Author SHA1 Message Date
Codex 0ff6e687b6 fix(security): ignorer les placeholders URL documentaires 2026-09-12 15:17:54 +02:00
Codex 573705a3d0 chore(integration): intégrer les PR Pronote critiques dans main
Revue Codex validée. Intégration dans main de la chaîne des PR #31, #32 et #33.

Co-authored-by: Codex <codex@antoineve.me>
Co-committed-by: Codex <codex@antoineve.me>
2026-09-12 15:00:32 +02:00
Codex d45d38d365 fix(ical): fiabiliser le parsing Pronote 2026
Revue Codex validée. Corrections de parsing iCal, conservation du contexte des devoirs et fixtures anonymisées.

Co-authored-by: Codex <codex@antoineve.me>
Co-committed-by: Codex <codex@antoineve.me>
2026-09-12 14:54:55 +02:00
AntoineVe f194ed985d fix(xmpp): corriger le timeout XMPP — TLS mode + timeouts de connexion/cleanup (#25) (#26) 2026-09-11 17:25:21 +02:00
AntoineVe d124f78b55 fix(dry-run) : garantir une simulation sans persistance (#24) 2026-09-11 12:00:40 +02:00
Codex 0dd4ee68c1 docs: define strict dry-run contract
# Conflicts:
#	.secrets.baseline
2026-09-11 11:57:21 +02:00
Codex fde8fbe264 fix: reject QR token dry-run 2026-09-11 11:44:13 +02:00
Codex 145dace289 fix: wire dry-run source state policy 2026-09-11 11:44:13 +02:00
Codex 642bbf16c1 fix: prevent source state writes in dry-run 2026-09-11 11:44:13 +02:00
AntoineVe 2b82ffb38d fix(auth) : sérialiser le cycle QR/token local (#23) 2026-09-11 11:41:57 +02:00
Codex 4228c1e636 fix: ignore QR token lockfile 2026-09-10 23:41:19 +02:00
Codex 8b924b55d1 docs: document QR token state locking 2026-09-10 21:26:43 +02:00
Codex 22a662ab39 fix: serialize QR token client operations 2026-09-10 21:24:11 +02:00
Codex 5188761209 fix: add QR token state lock primitive 2026-09-10 21:15:08 +02:00
37 changed files with 2560 additions and 330 deletions
+5 -1
View File
@@ -50,7 +50,11 @@ XMPP_HOST=example.com
XMPP_PORT=5222
XMPP_TO=parent@example.com
XMPP_RESOURCE=pronote-sync
XMPP_USE_TLS=true
# Mode TLS : "starttls" (port 5222, défaut), "direct" (port 5223) ou "disabled" (localhost uniquement)
XMPP_TLS_MODE=starttls
# XMPP_USE_TLS=true/false est obsolète mais encore supporté (alias de XMPP_TLS_MODE)
XMPP_CONNECT_TIMEOUT=15
XMPP_CLEANUP_TIMEOUT=10
XMPP_TIMEOUT=30
# --- IA (optionnelle) ---
+1
View File
@@ -50,6 +50,7 @@ Thumbs.db
.caldav_sync_state.json
# État d'authentification pronotepy (QR code / token rotation)
.pronote_auth_state.json
.pronote_auth_state.json.lock
*.state.json
# --- Local scratch / WIP files ---
+12 -4
View File
@@ -139,9 +139,8 @@
"type": "Hex High Entropy String",
"filename": "GUIDE_DEV_PYTHON.md",
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
"is_verified": true,
"line_number": 5064,
"is_secret": false
"is_verified": false,
"line_number": 5125
}
],
"tests/unit/test_caldav_gateway.py": [
@@ -175,7 +174,16 @@
"is_verified": false,
"line_number": 28
}
],
"tests/unit/test_xmpp_channel.py": [
{
"type": "Secret Keyword",
"filename": "tests/unit/test_xmpp_channel.py",
"hashed_secret": "b01bcded8826562355b3e32a8b31e214ff533511",
"is_verified": false,
"line_number": 1011
}
]
},
"generated_at": "2026-09-08T10:45:46Z"
"generated_at": "2026-09-12T12:04:07Z"
}
+78 -17
View File
@@ -275,7 +275,8 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati
> ⚠️ **Décision d'implémentation** :
> `XMPP_RECIPIENT` a été renommé en `XMPP_TO` dans l'implémentation (aligné avec §10.2.1).
> Des variables XMPP supplémentaires ont été ajoutées : `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_RESOURCE`, `XMPP_USE_TLS`, `XMPP_TIMEOUT`.
> Des variables XMPP supplémentaires ont été ajoutées : `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_RESOURCE`, `XMPP_TLS_MODE`, `XMPP_TIMEOUT`, `XMPP_CONNECT_TIMEOUT`, `XMPP_CLEANUP_TIMEOUT`.
> `XMPP_USE_TLS` reste supporté mais est **obsolète** (remplacé par `XMPP_TLS_MODE`).
> Une section `BLOG_ENABLED` et `BLOG_RSS_URL` a été ajoutée dans `.env.example`.
Les variables Pronote sont obligatoires selon les sources activées :
@@ -310,7 +311,7 @@ d'un besoin réel et testé.
| `AI_API_KEY` | Clé API pour l'API IA. | `None` | `SecretStr` |
| `AI_MODEL` | Modèle IA à utiliser (exemple recommandé : `gpt-4o-mini`). | `None` | `str \| None`|
| `AI_ALLOW_INSECURE_HTTP` | Autoriser HTTP (non sécurisé) pour `openai-compatible` uniquement. | `False` | `bool` |
| `DRY_RUN` | Mode dry-run (pas de modifications CalDAV/XMPP). | `False` | `bool` |
| `DRY_RUN` | Simulation sans sortie distante ni état local persistant ; incompatible avec `qr_token`. | `False` | `bool` |
| `LOG_LEVEL` | Niveau de log (`DEBUG`, `INFO`, `WARNING`, `ERROR`). | `INFO` | `str` |
> ⚠️ **Décision d'implémentation** :
@@ -1531,6 +1532,14 @@ Sinon :
retourner J+1 (no-school, avec libellé de vacances si applicable)
```
Dans l'implémentation Python actuelle (`pronote_sync.pipeline.steps.fetch.resolve_target_date`),
seuls les cours non annulés sont considérés comme effectifs ; un cours déplacé reste donc
éligible. La recherche du prochain cours connu ne dépend pas de la présence d'un cours à J.
Lorsqu'aucun cours futur n'est fourni et qu'un événement scolaire couvre J+1, sa borne
`to_date` (exclusive) est utilisée comme reprise connue. Sinon, J+1 est conservé comme
valeur de repli explicite. La fonction journalise la règle choisie avec la date et une raison
non personnelle.
**Exemple Python** :
```python
Optional, Tuple, List
@@ -1892,6 +1901,20 @@ def generate_deterministic_uid(
#### 5.1.6 Parsing complet du flux iCal (`sources/pronote/ical.py`)
Le parseur de production accepte les variantes contrôlées des libellés et des
balises `strong` générées par Pronote. Il distingue les catégories `Congés` /
`Vacances` (`HOLIDAY`) des catégories `Jour(s) férié(s)` (`PUBLIC_HOLIDAY`),
conserve la borne `DTEND` comme date exclusive et reconnaît les statuts
annulé, déplacé ou modifié. Les sections de devoirs conservent un HTML
nettoyé (scripts, styles et attributs exécutables supprimés) ainsi qu'un texte
lisible. La déduplication utilise la date, la matière, les enseignants et le
texte normalisé : une copie du même devoir est fusionnée, tandis qu'un devoir
homonyme d'une autre matière ou d'un autre enseignant est conservé.
Le champ `Partie(s) de classe` est conservé dans `Lesson.class_part` et dans
la description iCalendar sérialisée. Les catégories ou sections inconnues ne
doivent pas être interprétées comme des vacances, une annulation ou un devoir.
```python
List, Optional, Tuple
from datetime import datetime, date
@@ -2143,9 +2166,9 @@ def parse_ical(raw_ical: str) -> tuple[List[Lesson], List[HomeworkModel], List[S
- school_events : Liste des événements scolaires (vacances).
**Note importante** :
La déduplication globale des devoirs est effectuée **après le parsing** de tous les VEVENT,
une fois que `target_date` est connu (via `resolve_target_day`).
Voir la section [5.1.4 Déduplication des devoirs](#514-déduplication-des-devoirs) pour plus de détails.
La déduplication globale des devoirs est effectuée **après le parsing** de tous les VEVENT,
une fois que `target_date` est connu (via `resolve_target_day`).
Voir la section [5.1.4 Déduplication des devoirs](#514-déduplication-des-devoirs) pour plus de détails.
"""
cal = Calendar.from_ical(raw_ical)
@@ -2259,7 +2282,27 @@ d'informations sont non critiques et peuvent retourner une liste vide avec un wa
Les objets renvoyés par `client.homework(start, end)` couvrent une fenêtre. Le résultat destiné à
un jour cible est donc filtré explicitement sur `homework.date == target_date`.
#### 5.1.8 Logique de repli (`sources/pronote/fallback.py`)
#### 5.1.8 Verrou du cycle d'authentification QR/token
En mode `qr_token`, le token Pronote est un état partagé et rotatif. Afin d'éviter que deux
exécutions ne réutilisent ou n'écrasent cet état simultanément, le client protège chaque cycle
d'authentification et de récupération par un verrou POSIX local non bloquant, situé dans
`.pronote_auth_state.json.lock`, à côté de `.pronote_auth_state.json`.
Le verrou couvre l'ensemble du cycle QR/token : chargement de l'état, connexion par token ou
enrôlement QR initial, opération de données (agenda, devoirs, messages ou informations), puis
persistance des credentials actualisées. Une tentative concurrente échoue immédiatement avec une
erreur d'état d'authentification expurgée ; elle ne patiente pas et ne relance pas
l'authentification. Le contenu du token, le PIN et les autres credentials ne sont jamais inclus
dans les logs ni dans ce message d'erreur.
Ce mécanisme est un contrat **local** : il coordonne des processus sur le même hôte Linux et un
filesystem local. Pour des déploiements conteneurisés, les conteneurs qui partagent le même compte
Pronote doivent également partager le fichier d'état et son fichier de verrou. Le verrou ne fournit
aucune exclusion fiable entre plusieurs hôtes ou via NFS ; dans ces cas, l'opérateur doit prévoir
une exclusion externe ou utiliser un token distinct par instance.
#### 5.1.9 Logique de repli (`sources/pronote/fallback.py`)
Le `PronoteFetcher` dépend de `Settings` et d'un protocole de client injecté ; il ne construit pas
de singleton et ne contient pas d'identifiants dupliqués.
@@ -3221,7 +3264,8 @@ class CalDAVClient:
### 7.4 Points clés
- **Différentielle** : La synchronisation compare les UID existants avec ceux à synchroniser.
- **Idempotence** : Deux exécutions identiques ne modifient pas le calendrier.
- **Dry-run** : Mode obligatoire pour tester sans effet de bord.
- **Dry-run** : Les lectures sont autorisées, sans sortie distante ni état local persistant ; le
mode `qr_token` est refusé avant connexion car son authentification implique une rotation distante.
- **Marquage** : Les événements gérés sont marqués avec `X-PRONOTE-SYNC-MANAGED: v1` pour éviter les conflits.
- **Cours annulés** : Conservés avec `STATUS:CANCELLED` (ne pas supprimer).
- **Plan explicite** : Le `CalDAVSyncPlan` est calculé avant l'exécution.
@@ -4045,7 +4089,9 @@ Si le besoin évolue (ex: **plusieurs destinataires**), les étapes suivantes so
| `XMPP_PORT` | Port XMPP (5222 pour STARTTLS, 5223 pour TLS direct). | `5222` | `int` | ❌ Non |
| `XMPP_TO` | Destinataire unique (ex: `parent@exemple.org`). | `None` | `str` | ✅ Oui |
| `XMPP_RESOURCE` | Ressource XMPP (ex: `pronote-sync`). | `"pronote-sync"` | `str` | ❌ Non |
| `XMPP_USE_TLS` | Utiliser TLS pour la connexion. | `True` | `bool` | ❌ Non |
| `XMPP_TLS_MODE` | Mode TLS : `starttls` (défaut), `direct` ou `disabled` (localhost uniquement). | `"starttls"` | `Literal` | ❌ Non |
| `XMPP_CONNECT_TIMEOUT` | Timeout de connexion (secondes). | `15` | `float` | ❌ Non |
| `XMPP_CLEANUP_TIMEOUT` | Timeout de nettoyage (secondes). | `10` | `float` | ❌ Non |
| `XMPP_TIMEOUT` | Timeout de connexion (secondes). | `30` | `int` | ❌ Non |
**⚠️ Notes** :
@@ -4053,6 +4099,7 @@ Si le besoin évolue (ex: **plusieurs destinataires**), les étapes suivantes so
- **Pas de variables PubSub** : `XMPP_PUBSUB_NODE`, `XMPP_ROOM`, `XMPP_SUBSCRIBERS` **ne doivent pas être introduites** pour l'instant.
- **Sécurité** : `XMPP_JID`, `XMPP_PASSWORD` et `XMPP_TO` **ne doivent jamais apparaître** dans les logs, erreurs ou fixtures.
- **Standardisation** : `XMPP_TO` est mappé sur le champ `to` dans le modèle Pydantic.
- **`XMPP_USE_TLS` obsolète** : remplacée par `XMPP_TLS_MODE` (`starttls`, `direct` ou `disabled`) ; `XMPP_USE_TLS=true``direct` et `XMPP_USE_TLS=false``starttls`, avec un `DeprecationWarning`.
#### 10.2.2 Exemple de configuration dans `.env`
@@ -4065,13 +4112,19 @@ XMPP_HOST=exemple.org
XMPP_PORT=5222
XMPP_TO=parent@exemple.org
XMPP_RESOURCE=pronote-digest
XMPP_USE_TLS=true
# XMPP_TLS_MODE : "starttls" (défaut), "direct" (port 5223) ou "disabled" (localhost uniquement)
XMPP_TLS_MODE=starttls
# XMPP_USE_TLS=true/false est obsolète mais encore supporté (alias de XMPP_TLS_MODE)
XMPP_CONNECT_TIMEOUT=15
XMPP_CLEANUP_TIMEOUT=10
XMPP_TIMEOUT=30
```
#### 10.2.3 Modèle Pydantic pour la configuration XMPP
```python
from typing import Literal
from pydantic import SecretStr, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -4097,14 +4150,22 @@ class XmppSettings(BaseSettings):
port: int = Field(default=5222, ge=1, le=65535)
to: str | None = None
resource: str = "pronote-sync"
use_tls: bool = True
timeout: int = Field(default=30, gt=0)
tls_mode: Literal["direct", "starttls", "disabled"] = "starttls"
use_tls: bool | None = Field(
default=None,
deprecated="Utiliser tls_mode à la place (XMPP_USE_TLS obsolète).",
)
timeout: float = Field(default=30, gt=0)
connect_timeout: float = Field(default=15, gt=0)
cleanup_timeout: float = Field(default=10, gt=0)
```
> **⚠️ Mapping TLS** :
> - `use_tls=True`**TLS direct** (port 5223, `enable_direct_tls=True`, `enable_starttls=False`).
> - `use_tls=False`**STARTTLS** (port 5222, `enable_starttls=True`, `enable_direct_tls=False`).
> La validation refuse `use_tls=False` si `host` n'est pas un hôte de boucle locale (`localhost`, `127.0.0.1`, `::1`).
> - `tls_mode="starttls"` (défaut) → **STARTTLS** (port 5222, `enable_starttls=True`, `enable_direct_tls=False`).
> - `tls_mode="direct"`**TLS direct** (port 5223, `enable_direct_tls=True`, `enable_starttls=False`).
> - `tls_mode="disabled"` → TLS désactivé ; **uniquement** autorisé sur un hôte de boucle locale (`localhost`, `127.0.0.1`, `::1`).
> - `XMPP_USE_TLS` (`use_tls`) est **obsolète** : `use_tls=True``tls_mode="direct"` et
> `use_tls=False``tls_mode="starttls"` (avec `DeprecationWarning`).
---
@@ -5760,7 +5821,7 @@ repos:
|----------------------------------------|-----------------------------------------------------------------------------------------------------|-----------------|------------|
| Configuration des variables d'environnement | Vérifier que toutes les variables obligatoires sont définies (voir [Section 3.1](#31-variables-denvironnement)). | ✅ Oui | |
| Vérification des secrets | Exécuter le script de vérification de sécurité (voir [Section 13.6](#136-exemple-de-script-de-vérification-de-sécurité)). | ✅ Oui | |
| Test en mode dry-run | Exécuter le pipeline avec `DRY_RUN=true` pour vérifier que tout fonctionne sans effet de bord. | ✅ Oui | |
| Test en mode dry-run | Exécuter le pipeline avec `DRY_RUN=true` sans sortie distante ni état local persistant (`qr_token` exclu). | ✅ Oui | |
| Configuration des logs | Vérifier que les logs sont configurés avec masquage des secrets (voir [Section 4.2](#42-implémentation)). | ✅ Oui | |
| Vérification des dépendances | Exécuter `pip check` pour vérifier que toutes les dépendances sont installées. | ✅ Oui | |
| Configuration du cron (si planifié) | Configurer une tâche cron pour exécuter le script régulièrement (ex: tous les jours à 18h). | ⚠️ Non | |
@@ -5940,7 +6001,7 @@ Exemple de ligne cron (exécution tous les jours à 18h) :
| **CalDAV** | Protocole pour synchroniser des calendriers via HTTP. |
| **XMPP** | Protocole de messagerie instantanée (anciennement Jabber). |
| **UID** | Identifiant unique pour un événement iCal/CalDAV. |
| **Dry-run** | Mode de test où aucune modification n'est appliquée (lecture seule). |
| **Dry-run** | Simulation avec lectures autorisées, sans sortie distante ni état local persistant ; incompatible avec `qr_token`. |
| **Idempotence** | Propriété d'une opération qui produit le même résultat si elle est exécutée plusieurs fois. |
| **Reverse-engineering** | Technique consistant à analyser un logiciel pour en comprendre le fonctionnement interne. |
@@ -6107,7 +6168,7 @@ Ce guide fournit une **base architecturale et technique solide** pour développe
| **CalDAV** | Protocole pour synchroniser des calendriers via HTTP. |
| **XMPP** | Protocole de messagerie instantanée (anciennement Jabber). |
| **UID** | Identifiant unique pour un événement iCal/CalDAV. |
| **Dry-run** | Mode de test où aucune modification n'est appliquée (lecture seule). |
| **Dry-run** | Simulation avec lectures autorisées, sans sortie distante ni état local persistant ; incompatible avec `qr_token`. |
| **Idempotence** | Propriété d'une opération qui produit le même résultat si elle est exécutée plusieurs fois. |
| **Reverse-engineering** | Technique consistant à analyser un logiciel pour en comprendre le fonctionnement interne. |
+2 -1
View File
@@ -71,7 +71,8 @@ The following variables can be safely pre-configured in `/etc/pronote-sync/prono
- `THEORETICAL_WEEK_ANCHOR_DATE`, `THEORETICAL_WEEK_ANCHOR_TYPE`
- **XMPP:**
- `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_USE_TLS`, `XMPP_TIMEOUT`, `XMPP_RESOURCE`
- `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_TLS_MODE`, `XMPP_TIMEOUT`, `XMPP_CONNECT_TIMEOUT`, `XMPP_CLEANUP_TIMEOUT`, `XMPP_RESOURCE`
- `XMPP_USE_TLS` is deprecated but still supported (aliased to `XMPP_TLS_MODE`)
- **AI:**
- `AI_ENABLED`, `AI_PROVIDER`, `AI_BASE_URL`, `AI_MODEL`, `AI_ALLOW_INSECURE_HTTP`
+7 -1
View File
@@ -36,10 +36,16 @@ pronote-sync --dry-run --log-level DEBUG
```bash
pronote-sync # Exécute la synchronisation
pronote-sync --dry-run # Simulation sans écriture
pronote-sync --dry-run # Simulation : lectures autorisées, aucune écriture persistante ni sortie distante
pronote-sync --log-level DEBUG # Verbosité des journaux
```
En `--dry-run`, les données peuvent être lues pour construire la simulation, mais aucun état local
de source n'est enregistré : l'état RSS reste en mémoire pendant l'exécution. Les écritures CalDAV
et l'envoi XMPP sont également désactivés. Le mode
`PRONOTE_AUTH_MODE=qr_token` est refusé avant toute connexion, car la rotation de son token ne peut
pas garantir un état persistant cohérent pendant une simulation.
---
## 🛠️ Déploiement
+10 -4
View File
@@ -147,7 +147,8 @@ Synchroniser différentiellement les événements Pronote vers le calendrier Cal
- Le plan de sync est correctement calculé (données Pronote vs événements distants gérés).
- Un changement de source iCal ↔ `pronotepy` ne crée ni doublon ni suppression/ajout artificiel pour un cours équivalent.
- Un run dry-run n'écrit rien ; deux runs identiques donnent un résultat identique.
- Un run dry-run autorise les lectures, mais ne produit aucune sortie distante ni écriture locale
persistante ; `qr_token` est refusé avant connexion. Deux runs identiques donnent le même résultat.
- Les événements annulés restent (`STATUS:CANCELLED`) et sont marqués `MANAGED`.
- Les événements non marqués ne sont jamais modifiés ni supprimés.
@@ -223,7 +224,9 @@ Composer et orchestrer toutes les étapes avec gestion d'erreurs dégradée et m
- [x] Créer les étapes `pipeline/steps/` : `fetch.py`, `normalize.py`, `compare.py`, `caldav_sync.py`, `synthesis.py`, `send.py`, `fetch_blog.py`.
- [x] Créer `pipeline/run.py` : `PipelineRunner` (composition root) orchestrant fetch → normalize → fetch_blog → compare → caldav_sync → synthesis → send.
- [x] Gérer les erreurs dégradées (continuer sauf critique) et renvoyer `(PronoteData, erreurs + warns)`.
- [x] Implémenter le mode `dry_run` (aucune écriture CalDAV/XMPP).
- [x] Implémenter le mode `dry_run` : lectures autorisées, aucune écriture CalDAV/XMPP ni écriture
persistante locale ; l'état RSS reste en mémoire pendant l'exécution, et
`PRONOTE_AUTH_MODE=qr_token` est refusé avant toute connexion.
- [x] Câbler l'injection des dépendances (Protocol + composition root), sans singleton global.
- [x] Réutiliser, dans une même exécution, un unique téléchargement/parsing iCal pour l'agenda et les devoirs lorsque les sources sélectionnées le permettent ; rester sur un cache local au run, sans cache global ni persistant.
@@ -231,7 +234,9 @@ Composer et orchestrer toutes les étapes avec gestion d'erreurs dégradée et m
- [x] Le pipeline complet s'exécute de bout en bout (mocks) dans le bon ordre.
- [x] Une sélection iCal commune à l'agenda et aux devoirs ne déclenche qu'un téléchargement/parsing du flux par run.
- [x] Une erreur non critique (ex : synthèse IA) n'empêche pas l'envoi XMPP.
- [x] `dry_run=True` n'effectue aucune écriture ; aucune source disponible → erreur critique explicite.
- [x] `dry_run=True` autorise les lectures mais n'effectue aucune écriture persistante locale ni
sortie distante ; aucune source disponible → erreur critique explicite. L'état RSS n'est pas
enregistré et `qr_token` est refusé avant toute connexion.
- [x] Si `THEORETICAL_AGENDA_PATH` est absent, le pipeline produit un diff vide sans erreur et n'instancie pas `AgendaComparator` ; si présent, il instancie le comparateur et effectue la comparaison.
- [x] Les erreurs critiques (`PipelineCriticalError`) propagées depuis une étape non-bloquante arrêtent le pipeline.
@@ -247,7 +252,8 @@ Exposer le lancement du pipeline via une interface en ligne de commande.
- [x] Gérer le code de retour et l'affichage des erreurs (redactées).
### Critères d'acceptation
- `pronote-sync --dry-run --log-level DEBUG` s'exécute sans effet de bord.
- `pronote-sync --dry-run --log-level DEBUG` s'exécute sans sortie distante ni état local persistant ;
le mode `qr_token`, qui implique une rotation distante, est refusé avant connexion.
- Le script console est installable (`[project.scripts]` dans `pyproject.toml`).
- Les erreurs affichées ne contiennent aucun secret, y compris avec l'affichage d'un traceback complet en mode debug.
+5 -2
View File
@@ -42,8 +42,11 @@ bloquent donc pas le déploiement. Il ne valide ni les valeurs ni les permission
du fichier d'environnement. Pour analyser seulement le contenu indexé avant un
commit, utilisez `scripts/check_secrets.py --staged`.
Le dry-run vérifie le pipeline sans appliquer les écritures de synchronisation ;
il ne remplace pas une vérification des paramètres réellement chargés.
Le dry-run autorise les lectures nécessaires à la simulation, mais n'applique aucune sortie
CalDAV/XMPP et ne modifie aucun état local persistant. L'état RSS reste limité à la mémoire du
processus. Le mode `PRONOTE_AUTH_MODE=qr_token` est incompatible avec cette garantie : la commande
le refuse avant toute connexion afin de ne pas désynchroniser le token local du token distant.
Le dry-run ne remplace pas une vérification des paramètres réellement chargés.
## Installation systemd
+16
View File
@@ -0,0 +1,16 @@
# Corpus de fixtures Pronote
Le corpus versionné de `tests/fixtures/` est limité à des exemples iCalendar
minimaux et anonymisés. Il est rejouable sans connexion Pronote et sert à
vérifier les modèles métier finaux, pas seulement l'absence d'exception.
Chaque fichier `*.ics` de variante possède un manifeste JSON homonyme. Le
manifeste indique la source, la version connue (ou `unknown`), les scénarios
couverts et les invariants à préserver. Les réponses API Pronote ne sont pas
stockées tant qu'elles ne peuvent pas être réduites sans conserver de données
personnelles ou de secret.
Avant d'ajouter un export : supprimer les URL, jetons, identifiants, noms,
établissement et contenu personnel ; réduire le document aux propriétés
nécessaires ; exécuter `scripts/check_secrets.py` ; puis ajouter un test qui
vérifie les modèles `Lesson`, `Homework` et `SchoolEvent` obtenus.
+162 -44
View File
@@ -36,6 +36,24 @@ logger = logging.getLogger(__name__)
__all__ = ["XmppChannel", "SyncXmppChannel", "XmppMessage"]
async def _cancel_pending(
*futures: asyncio.Future[bool],
) -> None:
"""Annule les futures/tâches encore en attente et supprime le bruit.
À appeler avant chaque retour anticipé de :meth:`XmppChannel.send_async`
afin qu'aucune tentative de connexion ne survive au retour de la méthode.
:param futures: Futures ou tâches à annuler (les déjà terminées sont
ignorées pour la cancellation mais attendues pour purger l'attente).
:rtype: None
"""
for future in futures:
if not future.done():
future.cancel()
await asyncio.gather(*futures, return_exceptions=True)
def _secret_values(settings: XmppSettings) -> tuple[SecretStr | str, ...]:
"""Rassemble les secrets du canal XMPP pour le masquage des logs.
@@ -220,11 +238,14 @@ class XmppChannel:
"""Exécute le flux asynchrone d'envoi XMPP (U2).
Connecte le client ``slixmpp`` avec un hôte et un port explicites,
configure TLS avant la connexion, puis attend l'un des événements
``session_start``, ``failed_auth`` ou ``disconnected`` sous un
timeout unique avant d'envoyer un message direct ``chat`` au
configure TLS avant la connexion selon ``tls_mode`` (``direct``,
``starttls`` ou ``disabled``), attend la connexion sous
``connect_timeout`` puis l'un des événements ``session_start``,
``failed_auth``, ``connection_failed`` ou ``disconnected`` sous
``timeout`` avant d'envoyer un message direct ``chat`` au
destinataire configuré. La déconnexion est garantie par un bloc
``try/finally``. Aucun secret n'est journalisé (SEC-XMPP-02).
``try/finally`` borné par ``cleanup_timeout``. Aucun secret n'est
journalisé (SEC-XMPP-02).
:param message: Message final à envoyer.
:return: ``True`` si l'envoi a réussi (ou a été simulé en dry-run),
@@ -244,55 +265,132 @@ class XmppChannel:
logger.warning("Destinataire XMPP manquant.")
return False
# Create typed client
client = ClientXMPP(
jid_str,
self.settings.password.get_secret_value() if self.settings.password else "",
)
# Configure TLS BEFORE connect
if self.settings.use_tls:
# TLS direct (port 5223 typically)
client.enable_direct_tls = True
client.enable_starttls = False
else:
# STARTTLS (port 5222 typically)
client.enable_starttls = True
client.enable_direct_tls = False
# Register handlers
# Client typed lazily: the construction is done inside the try block so that
# any error is caught and converted to ``False`` (channel contract: never raise)
client: ClientXMPP | None = None
session_future: asyncio.Future[bool] = asyncio.get_event_loop().create_future()
failure_kind = "disconnected"
# Declared before the ``try`` so the exception handlers (CancelledError and
# Exception) can cancel any task still pending from ``asyncio.wait()``
connect_future: asyncio.Future[bool] | None = None
session_task: asyncio.Future[bool] | None = None
def on_session_start(event: object) -> None:
if not session_future.done():
session_future.set_result(True)
def on_failed_auth(event: object) -> None:
nonlocal failure_kind
if not session_future.done():
failure_kind = "failed_auth"
session_future.set_result(False)
def on_connection_failed(event: object) -> None:
nonlocal failure_kind
if not session_future.done():
failure_kind = "connection_failed"
session_future.set_result(False)
def on_disconnected(event: object) -> None:
if not session_future.done():
session_future.set_result(False)
client.add_event_handler("session_start", on_session_start)
client.add_event_handler("failed_auth", on_failed_auth)
client.add_event_handler("disconnected", on_disconnected)
try:
# Create typed client
client = ClientXMPP(
jid_str,
self.settings.password.get_secret_value() if self.settings.password else "",
)
# Configure TLS BEFORE connect (canonical tls_mode)
match self.settings.tls_mode:
case "direct":
client.enable_direct_tls = True
client.enable_starttls = False
case "starttls":
client.enable_direct_tls = False
client.enable_starttls = True
case "disabled":
client.enable_direct_tls = False
client.enable_starttls = False
# Register handlers
client.add_event_handler("session_start", on_session_start)
client.add_event_handler("failed_auth", on_failed_auth)
client.add_event_handler("connection_failed", on_connection_failed)
client.add_event_handler("disconnected", on_disconnected)
# Connect with explicit host and port
connect_future = client.connect(self.settings.host, self.settings.port)
await connect_future # connect() returns a Future, not a coroutine
connect_future = asyncio.ensure_future(
client.connect(self.settings.host, self.settings.port)
)
# Wait for one of the three events under a single timeout
try:
success = await asyncio.wait_for(session_future, timeout=self.settings.timeout)
except TimeoutError:
logger.warning("Délai d'attente de session XMPP dépassé.")
return False
# Wrap the session future in a task so that cancelling pending tasks
# during the concurrent wait never cancels ``session_future`` itself
async def _await_session() -> bool:
return await session_future
if not success:
logger.warning("Échec d'authentification ou déconnexion XMPP.")
session_task = asyncio.ensure_future(_await_session())
# Wait for the connection and the session event concurrently, bounded by
# connect_timeout as the global time limit: a ``connection_failed`` event
# can thus trigger an early return before the connect timeout expires
done, _pending = await asyncio.wait(
{connect_future, session_task},
timeout=self.settings.connect_timeout,
return_when=asyncio.FIRST_COMPLETED,
)
if session_task in done:
if not session_future.result():
# ``connection_failed``/``failed_auth``/``disconnected`` fired
# before the connection was resolved: immediate failure (fail fast)
if failure_kind == "connection_failed":
logger.warning("Échec de connexion réseau XMPP.")
else:
logger.warning("Échec d'authentification ou déconnexion XMPP.")
await _cancel_pending(connect_future, session_task)
return False
# ``session_start`` fired: the connection succeeded even if the
# connect future is still pending; proceed to send the message
await _cancel_pending(connect_future, session_task)
elif connect_future in done:
# The connection resolved: surface a connect error (redacted) if any
if not connect_future.cancelled():
connect_exc = connect_future.exception()
if connect_exc is not None and isinstance(connect_exc, Exception):
logger.warning(
"Échec de connexion XMPP : %s",
redact_exception(
connect_exc,
extra_secrets=_secret_values(self.settings),
),
)
await _cancel_pending(connect_future, session_task)
return False
# Connection established: wait for a session event under ``timeout``
try:
success = await asyncio.wait_for(
asyncio.shield(session_future), timeout=self.settings.timeout
)
except TimeoutError:
logger.warning("Délai d'attente de session XMPP dépassé.")
await _cancel_pending(connect_future, session_task)
return False
if not success:
if failure_kind == "connection_failed":
logger.warning("Échec de connexion réseau XMPP.")
else:
logger.warning("Échec d'authentification ou déconnexion XMPP.")
await _cancel_pending(connect_future, session_task)
return False
else:
# connect_timeout expired: cancel everything and fail fast
logger.warning(
"Délai de connexion XMPP dépassé (%ss).", self.settings.connect_timeout
)
await _cancel_pending(connect_future, session_task)
return False
# Send the message
@@ -300,19 +398,39 @@ class XmppChannel:
client.send_message(mto=JID(self.settings.to), mbody=formatted, mtype="chat")
return True
except asyncio.CancelledError:
# Contrat du canal : toujours retourner un booléen, même en cas
# d'annulation de la tâche appelante (cleanup exécuté par le finally).
logger.debug("Envoi XMPP annulé avant la fin de l'opération.")
pending = [f for f in (connect_future, session_task) if f is not None]
if pending:
await _cancel_pending(*pending)
return False
except Exception as exc:
redacted = redact_exception(exc)
redacted = redact_exception(exc, extra_secrets=_secret_values(self.settings))
extra = _secret_values(self.settings)
logger.warning("Erreur XMPP: %s", redact_secrets(redacted, extra_secrets=extra))
pending = [f for f in (connect_future, session_task) if f is not None]
if pending:
await _cancel_pending(*pending)
return False
finally:
try:
disconnect_future = client.disconnect()
await disconnect_future
except Exception as cleanup_exc:
logger.debug(
"Erreur lors de la déconnexion XMPP: %s", redact_exception(cleanup_exc)
)
if client is not None:
try:
disconnect_future = client.disconnect()
await asyncio.wait_for(disconnect_future, timeout=self.settings.cleanup_timeout)
except asyncio.CancelledError:
logger.debug("Déconnexion XMPP annulée.")
except TimeoutError:
logger.debug(
"Délai de déconnexion XMPP dépassé (%ss), abandon.",
self.settings.cleanup_timeout,
)
except Exception as cleanup_exc:
logger.debug(
"Erreur lors de la déconnexion XMPP : %s",
redact_exception(cleanup_exc, extra_secrets=_secret_values(self.settings)),
)
class SyncXmppChannel:
@@ -363,7 +481,7 @@ class SyncXmppChannel:
try:
return asyncio.run(self._channel.send_async(message))
except Exception as exc:
redacted = redact_exception(exc)
redacted = redact_exception(exc, extra_secrets=_secret_values(self.settings))
redacted = redact_secrets(redacted, extra_secrets=_secret_values(self.settings))
logger.warning("XMPP : erreur lors de l'envoi synchrone : %s", redacted)
return False
+4 -1
View File
@@ -32,7 +32,10 @@ def _parse_arguments(arguments: Sequence[str] | None = None) -> argparse.Namespa
"--dry-run",
action="store_true",
default=None,
help="Simule la synchronisation sans écrire vers CalDAV ni XMPP.",
help=(
"Simule la synchronisation sans sortie distante ni état local persistant "
"(incompatible avec PRONOTE_AUTH_MODE=qr_token)."
),
)
parser.add_argument(
"--log-level",
+60 -24
View File
@@ -8,6 +8,7 @@ depuis les variables d'environnement (préfixées par groupe) et le fichier
from __future__ import annotations
import warnings
from datetime import date
from typing import Literal
from urllib.parse import urlparse
@@ -18,6 +19,7 @@ from pydantic import (
ValidationInfo,
field_serializer,
field_validator,
model_validator,
)
from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -158,10 +160,16 @@ class XmppSettings(BaseSettings):
Contraintes de champs : ``port`` est borné entre 1 et 65535 et ``timeout``
doit être strictement positif.
Politique TLS : la désactivation de TLS (``use_tls`` à ``False``) n'est
autorisée que sur un hôte de boucle locale (``localhost``, ``127.0.0.1``,
``::1``). Dans tout autre cas, une erreur de validation est levée,
indépendamment de l'état du champ ``enabled``.
Politique TLS : le mode ``tls_mode`` détermine la négociation TLS
(``direct``, ``starttls`` ou ``disabled``). Le mode ``disabled`` n'est
autorisé que sur un hôte de boucle locale (``localhost``, ``127.0.0.1``,
``::1``) ; ``starttls`` et ``direct`` sont permis pour tous les hôtes.
Compatibilité : le champ historique ``use_tls`` (booléen) est un alias
obsolète ; ``use_tls=True`` mappe vers ``tls_mode="direct"`` et
``use_tls=False`` vers ``tls_mode="starttls"``, avec un
:pyexc:`DeprecationWarning`. La valeur brute fournie reste lisible via
``use_tls`` (``None`` si non fournie).
"""
model_config = SettingsConfigDict(
@@ -178,33 +186,61 @@ class XmppSettings(BaseSettings):
port: int = Field(default=5222, ge=1, le=65535)
to: str | None = None
resource: str = "pronote-sync"
use_tls: bool = True
timeout: int = Field(default=30, gt=0)
tls_mode: Literal["direct", "starttls", "disabled"] = "starttls"
use_tls: bool | None = Field(
default=None,
deprecated="Utiliser tls_mode à la place (XMPP_USE_TLS obsolète).",
)
timeout: float = Field(default=30, gt=0)
connect_timeout: float = Field(default=15, gt=0)
cleanup_timeout: float = Field(default=10, gt=0)
@field_validator("use_tls")
@model_validator(mode="before")
@classmethod
def _validate_tls_policy(cls, v: bool, info: ValidationInfo) -> bool:
"""Refuse la désactivation de TLS hors des hôtes de boucle locale.
def _migrate_use_tls(cls, data: object) -> object:
"""Mappe l'alias obsolète ``use_tls`` vers le mode canonique ``tls_mode``.
La règle s'applique quel que soit l'état du champ ``enabled``. Le
message d'erreur ne contient aucune valeur sensible (``jid``,
``use_tls=True`` devient ``tls_mode="direct"`` et ``use_tls=False``
devient ``tls_mode="starttls"`` ; un :pyexc:`DeprecationWarning` est
émis à chaque usage explicite de l'alias. ``tls_mode`` fourni
explicitement prend le pas sur l'alias.
:param data: Données d'entrée du modèle (dict ou autre).
:return: Données d'entrée avec ``tls_mode`` dérivé de ``use_tls``.
:rtype: object
"""
if not isinstance(data, dict) or "use_tls" not in data:
return data
warnings.warn(
"XMPP_USE_TLS est obsolète : utiliser XMPP_TLS_MODE "
"('direct', 'starttls' ou 'disabled').",
DeprecationWarning,
stacklevel=2,
)
if data.get("tls_mode") is None:
data["tls_mode"] = "direct" if data["use_tls"] else "starttls"
return data
@model_validator(mode="after")
def _validate_tls_policy(self) -> XmppSettings:
"""Refuse le mode ``disabled`` hors des hôtes de boucle locale.
La règle s'applique quel que soit l'état du champ ``enabled``. Les
modes ``starttls`` et ``direct`` sont autorisés pour tous les hôtes.
Le message d'erreur ne contient aucune valeur sensible (``jid``,
``password``, ``to``).
:param v: Valeur du champ ``use_tls`` à valider.
:param info: Contexte de validation (accès aux autres champs).
:return: La valeur validée inchangée.
:rtype: bool
:raises ValueError: Si ``use_tls`` est ``False`` et que ``host``
:return: L'instance validée inchangée.
:rtype: XmppSettings
:raises ValueError: Si ``tls_mode`` est ``disabled`` et que ``host``
n'est pas un hôte de boucle locale.
"""
if v is False:
host = info.data.get("host", "")
if host not in _XMPP_LOOPBACK_HOSTS:
raise ValueError(
"TLS désactivé n'est autorisé que sur les hôtes de loopback "
"(localhost, 127.0.0.1, ::1)."
) from None
return v
if self.tls_mode == "disabled" and self.host not in _XMPP_LOOPBACK_HOSTS:
raise ValueError(
"TLS désactivé n'est autorisé que sur les hôtes de loopback "
"(localhost, 127.0.0.1, ::1)."
) from None
return self
class AISettings(BaseSettings):
+15
View File
@@ -38,6 +38,21 @@ class PronoteAuthRotationError(PronoteSyncError):
super().__init__(message)
class PronoteAuthStateLockError(PronoteSyncError):
"""Erreur levée lorsqu'un autre processus détient l'état d'authentification.
Cette erreur indique qu'une opération QR code / token concurrente est en
cours. Son message ne contient ni chemin local sensible ni credential.
"""
def __init__(self, message: str) -> None:
"""Initialise l'erreur de contention du verrou d'état.
:param message: Message actionnable expurgé décrivant la contention.
"""
super().__init__(message)
class ErrorSeverity(StrEnum):
"""Niveau de gravité d'une erreur produite par le pipeline."""
+1
View File
@@ -52,6 +52,7 @@ class Lesson(BaseModel):
teachers: tuple[str, ...] = Field(default=(), description="Liste des professeurs")
rooms: tuple[str, ...] = Field(default=(), description="Liste des salles")
group: str | None = Field(None, description="Groupe (ex: Classe entière)")
class_part: str | None = None
status: LessonStatus = Field(default=LessonStatus.NORMAL, description="Statut du cours")
content: str | None = Field(None, description="Contenu pédagogique")
homework_blocks: tuple[HomeworkBlock, ...] = Field(
+9 -2
View File
@@ -126,6 +126,9 @@ class PipelineRunner:
:rtype: PipelineRunner
"""
effective_dry_run = settings.app.dry_run if dry_run is None else dry_run
if effective_dry_run and settings.pronote.auth_mode == "qr_token":
raise ValueError("Le mode qr_token n'est pas compatible avec le dry-run.")
persistence_enabled = not effective_dry_run
theoretical_provider = get_theoretical_provider(
settings.app.theoretical_agenda_path,
settings.app.school_holidays_path,
@@ -136,7 +139,9 @@ class PipelineRunner:
AgendaComparator(theoretical_provider) if theoretical_provider is not None else None
)
blog_client = BlogRSSClient(settings.blog.rss_url) if settings.blog.enabled else None
blog_state = BlogRSSState() if settings.blog.enabled else None
blog_state = (
BlogRSSState(persistence_enabled=persistence_enabled) if settings.blog.enabled else None
)
return cls(
settings=settings,
pronote_fetcher=PronoteFetcher(
@@ -144,7 +149,9 @@ class PipelineRunner:
PronoteClient(
settings.pronote,
auth_state=(
PronoteAuthState() if settings.pronote.auth_mode == "qr_token" else None
PronoteAuthState(persistence_enabled=persistence_enabled)
if settings.pronote.auth_mode == "qr_token"
else None
),
),
),
+32 -15
View File
@@ -2,16 +2,19 @@
from __future__ import annotations
import logging
from dataclasses import dataclass
from datetime import date
from pronote_sync.errors import PipelineCriticalError, PipelineWarning, PronoteAuthRotationError
from pronote_sync.models.agenda import Lesson, SchoolEvent
from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent
from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message
from pronote_sync.sources.pronote.fallback import PronoteFetcherProtocol
from pronote_sync.utils.redaction import redact_exception
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class FetchedPronoteData:
@@ -36,27 +39,41 @@ def resolve_target_date(
) -> date:
"""Détermine la date cible du digest à partir de l'agenda disponible.
La règle privilégie J+1 lorsqu'il contient des cours. Si la journée en
cours contient des cours mais pas J+1, le prochain cours connu est choisi.
Sans cours correspondant, J+1 est conservé, y compris pendant les vacances.
La règle privilégie J+1 lorsqu'il contient un cours non annulé. Sinon,
elle choisit le prochain jour possédant un cours effectif dans la fenêtre.
Si aucune reprise n'est visible, la fin exclusive d'un événement scolaire
couvrant J+1 est utilisée ; à défaut, J+1 reste le meilleur choix connu.
:param today: Date de référence de l'exécution.
:param lessons: Cours récupérés pour la fenêtre de synchronisation.
:param school_events: Événements scolaires récupérés (réservés aux évolutions
du libellé de jour sans cours).
:param school_events: Événements scolaires récupérés, avec borne de fin exclusive.
:return: Date cible du digest.
:rtype: date
"""
del school_events
tomorrow = date.fromordinal(today.toordinal() + 1)
lesson_dates = {lesson.start.date() for lesson in lessons}
if tomorrow in lesson_dates:
return tomorrow
if today in lesson_dates:
future_dates = sorted(day for day in lesson_dates if day > today)
if future_dates:
return future_dates[0]
return tomorrow
effective_dates = {
lesson.start.date()
for lesson in lessons
if lesson.status != LessonStatus.CANCELLED and lesson.start.date() > today
}
if tomorrow in effective_dates:
reason = "cours effectif demain"
target = tomorrow
elif effective_dates:
reason = "prochain cours effectif connu"
target = min(effective_dates)
else:
event_ends = [
event.to_date for event in school_events if event.from_date <= tomorrow < event.to_date
]
if event_ends:
reason = "fin exclusive de l'événement couvrant demain"
target = max(event_ends)
else:
reason = "aucune reprise connue"
target = tomorrow
logger.info("Date cible %s : %s", target.isoformat(), reason)
return target
def _fetch_optional_messages(
+12 -1
View File
@@ -37,15 +37,24 @@ class BlogRSSState:
:param state_file: Chemin du fichier d'état JSON (``str`` ou
:class:`~pathlib.Path`). ``".blog_rss_state.json"`` par défaut.
:param persistence_enabled: Si ``False``, charge l'état existant mais ne
modifie jamais le fichier d'état. ``True`` par défaut.
"""
def __init__(self, state_file: Path | str = ".blog_rss_state.json") -> None:
def __init__(
self,
state_file: Path | str = ".blog_rss_state.json",
persistence_enabled: bool = True,
) -> None:
"""Initialise le gestionnaire d'état depuis le fichier JSON.
:param state_file: Chemin du fichier d'état JSON (``str`` ou
:class:`~pathlib.Path`). ``".blog_rss_state.json"`` par défaut.
:param persistence_enabled: Si ``False``, charge l'état existant mais
désactive toutes les écritures sur disque. ``True`` par défaut.
"""
self._state_file = Path(state_file)
self._persistence_enabled = persistence_enabled
self._known_guids: set[str] = set()
self._etag: str | None = None
self._last_modified: str | None = None
@@ -98,6 +107,8 @@ class BlogRSSState:
d'erreur d'écriture, une erreur est journalisée sans être
propagée et le fichier temporaire est supprimé.
"""
if not self._persistence_enabled:
return
payload = {
"version": _STATE_VERSION,
"known_guids": sorted(self._known_guids),
+95 -3
View File
@@ -14,10 +14,13 @@ from __future__ import annotations
import json
import logging
import os
from collections.abc import Generator
from contextlib import contextmanager
from fcntl import LOCK_EX, LOCK_NB, LOCK_UN, flock
from pathlib import Path
from typing import Any
from pronote_sync.errors import PronoteSyncError
from pronote_sync.errors import PronoteAuthStateLockError, PronoteSyncError
from pronote_sync.utils.redaction import redact_exception, redact_secrets
logger = logging.getLogger(__name__)
@@ -38,9 +41,15 @@ class PronoteAuthState:
:param state_file: Chemin du fichier d'état JSON (``str`` ou
:class:`~pathlib.Path`). ``".pronote_auth_state.json"`` par défaut.
:param persistence_enabled: Si ``False``, charge l'état existant mais ne
modifie jamais le fichier d'état. ``True`` par défaut.
"""
def __init__(self, state_file: Path | str = ".pronote_auth_state.json") -> None:
def __init__(
self,
state_file: Path | str = ".pronote_auth_state.json",
persistence_enabled: bool = True,
) -> None:
"""Initialise le gestionnaire d'état d'authentification Pronote.
Le fichier d'état n'est pas créé à l'initialisation : il n'est écrit
@@ -48,8 +57,14 @@ class PronoteAuthState:
:param state_file: Chemin du fichier d'état JSON (``str`` ou
:class:`~pathlib.Path`). ``".pronote_auth_state.json"`` par défaut.
:param persistence_enabled: Si ``False``, charge l'état existant mais
désactive toutes les écritures ou suppressions sur disque. ``True``
par défaut.
"""
self._state_file = Path(state_file)
self._persistence_enabled = persistence_enabled
self._in_memory_credentials: dict[str, str] | None = None
self._in_memory_state_cleared = False
def load(self) -> dict[str, str] | None:
"""Charge les credentials d'authentification depuis le fichier d'état.
@@ -65,6 +80,10 @@ class PronoteAuthState:
aucun état valide n'est disponible.
:rtype: dict[str, str] | None
"""
if self._in_memory_state_cleared:
return None
if self._in_memory_credentials is not None:
return self._in_memory_credentials.copy()
if not self._state_file.exists():
logger.debug(
"Fichier d'état d'authentification Pronote %s absent, aucun token à charger.",
@@ -108,6 +127,69 @@ class PronoteAuthState:
credentials[key] = value
return credentials
@contextmanager
def lock(self) -> Generator[None]:
"""Protège une opération d'état par un verrou POSIX non bloquant.
Le verrou est conservé dans le fichier frère ``<state_file>.lock`` afin
de survivre à l'écriture atomique du fichier d'état. Le fichier de
verrou reste présent après libération et est créé en ``0600`` pour ne
pas élargir l'accès aux métadonnées de l'état sensible.
:return: Un gestionnaire de contexte qui tient le verrou exclusif.
:rtype: collections.abc.Generator[None, None, None]
:raises PronoteAuthStateLockError: Si un autre processus détient déjà
le verrou ou si son acquisition échoue.
"""
lock_file = self._state_file.with_name(f"{self._state_file.name}.lock")
descriptor: int | None = None
try:
descriptor = os.open(
str(lock_file),
os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW,
0o600,
)
os.fchmod(descriptor, 0o600)
except OSError:
logger.error("Impossible d'ouvrir le verrou d'état d'authentification Pronote.")
if descriptor is not None:
os.close(descriptor)
if descriptor is None:
raise PronoteAuthStateLockError(
"Impossible d'acquérir le verrou d'état d'authentification Pronote."
) from None
is_contended = False
lock_acquisition_failed = False
try:
flock(descriptor, LOCK_EX | LOCK_NB)
except BlockingIOError:
is_contended = True
except OSError:
logger.error("Impossible d'acquérir le verrou d'état d'authentification Pronote.")
os.close(descriptor)
lock_acquisition_failed = True
if lock_acquisition_failed:
raise PronoteAuthStateLockError(
"Impossible d'acquérir le verrou d'état d'authentification Pronote."
) from None
if is_contended:
os.close(descriptor)
raise PronoteAuthStateLockError(
"Une autre opération d'authentification Pronote est déjà en cours."
)
try:
yield
finally:
try:
flock(descriptor, LOCK_UN)
finally:
os.close(descriptor)
def save(self, credentials: dict[str, str]) -> None:
"""Sauvegarde les credentials dans le fichier d'état, de manière atomique.
@@ -126,6 +208,10 @@ class PronoteAuthState:
:raises PronoteSyncError: Si l'écriture ou le remplacement du fichier
échoue.
"""
if not self._persistence_enabled:
self._in_memory_credentials = credentials.copy()
self._in_memory_state_cleared = False
return
payload: dict[str, Any] = {
"version": _STATE_VERSION,
"credentials": credentials,
@@ -182,10 +268,16 @@ class PronoteAuthState:
"""Supprime le fichier d'état d'authentification.
Si le fichier n'existe pas, la méthode ne fait rien et aucune erreur
n'est levée.
n'est levée. Lorsque la persistance est désactivée, elle efface
uniquement les credentials conservés en mémoire et ne modifie jamais
le fichier d'état.
:raises OSError: Si la suppression du fichier existant échoue.
"""
self._in_memory_credentials = None
if not self._persistence_enabled:
self._in_memory_state_cleared = True
return
if not self._state_file.exists():
return
logger.debug(
+131 -104
View File
@@ -12,6 +12,8 @@ from __future__ import annotations
import json
import logging
from collections.abc import Generator
from contextlib import contextmanager
from datetime import date
from pathlib import Path
from typing import Any, Protocol
@@ -23,10 +25,11 @@ import requests
from pronote_sync.config.settings import PronoteSettings
from pronote_sync.errors import PronoteAuthRotationError
from pronote_sync.models.agenda import Lesson, LessonStatus
from pronote_sync.models.agenda import Lesson
from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message, MessageType
from pronote_sync.sources.pronote.auth_state import PronoteAuthState
from pronote_sync.sources.pronote.lessons import collapse_replaced_lessons, lesson_status
from pronote_sync.utils.redaction import redact_exception, redact_secrets
from pronote_sync.utils.uid import generate_deterministic_uid, normalize_pronote_uid
@@ -252,6 +255,26 @@ class PronoteClient:
except Exception as exc:
logger.debug("Échec de la persistance des credentials : %s", redact_exception(exc))
@contextmanager
def _qr_token_operation_lock(self) -> Generator[None]:
"""Verrouille un cycle d'authentification et de récupération QR/token.
Le verrou englobe le chargement du token, le login, l'opération de
données et la persistance qui suit. Il est volontairement absent du
mode ``password``, qui ne partage pas de fichier d'état de token.
:return: Un gestionnaire de contexte protégeant le cycle QR/token.
:rtype: collections.abc.Generator[None, None, None]
:raises PronoteAuthStateLockError: Si l'état QR/token est déjà utilisé
par une autre opération.
"""
if self._settings.auth_mode != "qr_token" or self._auth_state is None:
yield
return
with self._auth_state.lock():
yield
def _connect_password(self) -> pronotepy.Client:
"""Connecte le client ``pronotepy`` en mode ``password``.
@@ -426,37 +449,38 @@ class PronoteClient:
: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,
with self._qr_token_operation_lock():
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,
)
)
)
self._persist_credentials()
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),
)
self._persist_credentials()
return []
self._persist_credentials()
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),
)
self._persist_credentials()
return []
def get_informations(self) -> list[Message]:
"""Récupère les informations et sondages Pronote.
@@ -467,36 +491,37 @@ class PronoteClient:
: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,
with self._qr_token_operation_lock():
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,
)
)
self._persist_credentials()
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),
)
self._persist_credentials()
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),
)
self._persist_credentials()
return []
self._persist_credentials()
return []
def get_lessons(self, start: date, end: date) -> list[Lesson]:
"""Récupère les cours via ``pronotepy`` (repli iCal).
@@ -523,37 +548,38 @@ class PronoteClient:
: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
raw_uid = lesson.id
if raw_uid:
uid = normalize_pronote_uid(raw_uid)
else:
uid = generate_deterministic_uid(
start=lesson.start,
end=lesson.end,
subject=lesson.subject.name if lesson.subject is not None else "",
teachers=list(lesson.teacher_names or ()),
rooms=list(lesson.classrooms or ()),
group=lesson.group_name,
with self._qr_token_operation_lock():
client = self._connect()
lessons: list[Lesson] = []
for lesson in client.lessons(start, end):
content = lesson.content
raw_uid = lesson.id
if raw_uid:
uid = normalize_pronote_uid(raw_uid)
else:
uid = generate_deterministic_uid(
start=lesson.start,
end=lesson.end,
subject=lesson.subject.name if lesson.subject is not None else "",
teachers=list(lesson.teacher_names or ()),
rooms=list(lesson.classrooms or ()),
group=lesson.group_name,
)
lessons.append(
Lesson(
id=uid,
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=lesson_status(getattr(lesson, "status", None), lesson.canceled),
content=content.description if content is not None else None,
)
)
lessons.append(
Lesson(
id=uid,
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,
)
)
self._persist_credentials()
return lessons
self._persist_credentials()
return collapse_replaced_lessons(lessons)
def get_homeworks(self, start: date, end: date) -> list[Homework]:
"""Récupère les devoirs via ``pronotepy``.
@@ -576,19 +602,20 @@ class PronoteClient:
: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,
with self._qr_token_operation_lock():
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,
)
)
)
self._persist_credentials()
return homeworks
self._persist_credentials()
return homeworks
+181 -76
View File
@@ -14,13 +14,15 @@ from __future__ import annotations
import hashlib
import re
import unicodedata
import urllib.parse
from datetime import date, datetime
from html import unescape
from pathlib import Path
from typing import TypedDict
from typing import NamedTuple, TypedDict
import requests
from bs4 import BeautifulSoup
from icalendar import Calendar
from ...models.agenda import (
@@ -33,22 +35,15 @@ from ...models.agenda import (
from ...models.homework import Homework
from ...utils.redaction import redact_exception, redact_url
from ...utils.uid import generate_deterministic_uid, normalize_pronote_uid
from .lessons import collapse_replaced_lessons
_HEADER_LABEL_PATTERN = re.compile(r"\b(Matière|Professeurs?|Salles?|Groupe)\s*:\s*")
_HEADER_LABEL_PATTERN = re.compile(
r"(?P<label>Mati(?:ère|ere)|Professeur(?:s|\(s\))?|Salle(?:s|\(s\))?"
r"|Groupe|Partie(?:s|\(s\))?\s+de\s+classe)\s*:\s*",
re.IGNORECASE,
)
_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,
)
_STRONG_PATTERN = re.compile(r"<strong\b[^>]*>(?P<label>.*?)</strong>", re.IGNORECASE | re.DOTALL)
_HEADERS = {
"accept": "text/calendar",
@@ -63,6 +58,15 @@ class HeaderInfo(TypedDict):
teachers: list[str]
rooms: list[str]
group: str | None
class_part: str | None
class ParsedHomeworkBlock(NamedTuple):
"""Bloc de devoir parsé avec son texte nettoyé et son HTML sûr."""
date: date
text: str
html: str
def fetch_ical(url: str, timeout: int = 20) -> str:
@@ -156,11 +160,11 @@ def split_header_and_body(description: str) -> tuple[str, str]:
: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:
strong_match = _STRONG_PATTERN.search(description)
if strong_match is None:
return description.strip(), ""
header = description[:strong_start].strip()
body = description[strong_start:]
header = description[: strong_match.start()].strip()
body = description[strong_match.start() :]
return header, body
@@ -168,7 +172,7 @@ 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
``Salle(s) :``, ``Groupe :`` et ``Partie(s) de classe :``. 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.
@@ -176,14 +180,21 @@ def parse_header(header: str) -> HeaderInfo:
: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))
info: HeaderInfo = {
"subject": "",
"teachers": [],
"rooms": [],
"group": None,
"class_part": None,
}
header_text = _strip_html(header)
matches = list(_HEADER_LABEL_PATTERN.finditer(header_text))
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":
value_end = matches[index + 1].start() if index + 1 < len(matches) else len(header_text)
value = unescape(header_text[value_start:value_end].strip())
label = _normalize_label(match.group("label")).replace("(s)", "s")
if label == "matiere":
info["subject"] = value
elif label in ("professeur", "professeurs"):
info["teachers"] = [part.strip() for part in value.split(",") if part.strip()]
@@ -191,9 +202,44 @@ def parse_header(header: str) -> HeaderInfo:
info["rooms"] = [part.strip() for part in value.split(",") if part.strip()]
elif label == "groupe":
info["group"] = value
elif label in ("partie de classe", "parties de classe"):
info["class_part"] = value
return info
def _normalize_label(value: str) -> str:
"""Normalise un libellé iCal pour comparer des variantes contrôlées.
:param value: Libellé à normaliser.
:return: Libellé minuscule sans accents et avec des espaces unifiés.
:rtype: str
"""
decomposed = unicodedata.normalize("NFKD", value)
without_accents = "".join(char for char in decomposed if not unicodedata.combining(char))
return re.sub(r"\s+", " ", without_accents).strip().lower()
def _sanitize_html(fragment: str) -> str:
"""Nettoie un fragment HTML de description sans exécuter de contenu.
:param fragment: Fragment HTML extrait d'une section de devoir.
:return: HTML conservé sans scripts, styles ni attributs exécutables.
:rtype: str
"""
soup = BeautifulSoup(fragment, "html.parser")
for tag in soup.find_all(("script", "style")):
tag.decompose()
for tag in soup.find_all(True):
for attribute in list(tag.attrs):
lowered = attribute.lower()
value = tag.attrs[attribute]
if lowered.startswith("on") or (
lowered in ("href", "src") and str(value).lower().strip().startswith("javascript:")
):
del tag.attrs[attribute]
return soup.decode_contents().strip()
def _strip_html(text: str) -> str:
"""Retire les balises HTML d'un texte et nettoie les espaces.
@@ -201,8 +247,10 @@ def _strip_html(text: str) -> str:
:return: Texte brut sans balises, entités HTML décodées.
:rtype: str
"""
cleaned = _TAG_PATTERN.sub("", text)
return unescape(cleaned).strip()
soup = BeautifulSoup(text, "html.parser")
for tag in soup.find_all(("script", "style")):
tag.decompose()
return " ".join(unescape(soup.get_text(" ", strip=True)).split())
def _parse_french_date(value: str) -> date | None:
@@ -218,44 +266,57 @@ def _parse_french_date(value: str) -> date | None:
return None
def parse_body(body: str) -> tuple[str | None, list[tuple[date, str]], list[tuple[date, str]]]:
def parse_body(
body: str,
) -> tuple[str | None, list[ParsedHomeworkBlock], list[ParsedHomeworkBlock]]:
"""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.
et les devoirs donnés des sections ``<strong>Donné le JJ/MM/AAAA :</strong>``.
Les listes préservent tous les blocs, même lorsque plusieurs sections partagent
la même date, avec le texte nettoyé et le HTML sûr de chaque bloc.
: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]]]
:rtype: tuple[str | None, list[ParsedHomeworkBlock], list[ParsedHomeworkBlock]]
"""
content: str | None = None
due_blocks: list[tuple[date, str]] = []
assigned_blocks: list[tuple[date, str]] = []
due_blocks: list[ParsedHomeworkBlock] = []
assigned_blocks: list[ParsedHomeworkBlock] = []
content_match = _CONTENT_PATTERN.search(body)
if content_match is not None:
content = _strip_html(content_match.group(1))
matches = list(_STRONG_PATTERN.finditer(body))
for index, match in enumerate(matches):
next_start = matches[index + 1].start() if index + 1 < len(matches) else len(body)
heading = _strip_html(match.group("label")).rstrip(":").strip()
fragment = body[match.end() : next_start]
safe_html = _sanitize_html(fragment)
text = _strip_html(fragment)
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))))
normalized_heading = _normalize_label(heading)
if normalized_heading == "contenu pedagogique":
content = text
continue
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))))
due_match = re.fullmatch(r"Pour\s+le\s+(\d{2}/\d{2}/\d{4})", heading, re.IGNORECASE)
assigned_match = re.fullmatch(
r"Donne\s+le\s+(\d{2}/\d{2}/\d{4})", normalized_heading, re.IGNORECASE
)
if due_match is not None:
due_date = _parse_french_date(due_match.group(1))
if due_date is not None:
due_blocks.append(ParsedHomeworkBlock(due_date, text, safe_html))
elif assigned_match is not None:
assigned_date = _parse_french_date(assigned_match.group(1))
if assigned_date is not None:
assigned_blocks.append(ParsedHomeworkBlock(assigned_date, text, safe_html))
return content, due_blocks, assigned_blocks
def parse_homework_blocks(
due_blocks: list[tuple[date, str]],
assigned_blocks: list[tuple[date, str]],
due_blocks: list[ParsedHomeworkBlock],
assigned_blocks: list[ParsedHomeworkBlock],
) -> tuple[HomeworkBlock, ...]:
"""Construit les :class:`HomeworkBlock` depuis les listes de devoirs.
@@ -269,10 +330,12 @@ def parse_homework_blocks(
: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))
for block in due_blocks:
blocks.append(HomeworkBlock(kind="due", date=block.date, text=block.text, html=block.html))
for block in assigned_blocks:
blocks.append(
HomeworkBlock(kind="assigned", date=block.date, text=block.text, html=block.html)
)
return tuple(blocks)
@@ -286,24 +349,32 @@ def normalize_homework_text(text: str) -> str:
:return: Texte normalisé.
:rtype: str
"""
normalized = re.sub(r"\s+", " ", text)
normalized = _TAG_PATTERN.sub("", normalized)
return normalized.strip().lower()
return _strip_html(text).casefold()
def generate_homework_id(due_on: date, normalized_text: str) -> str:
def generate_homework_id(
due_on: date,
normalized_text: str,
subject: str = "",
teachers: tuple[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
L'ID est la clé ``AAAA-MM-JJ|matière|enseignants|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.
:param subject: Matière du devoir, utile pour distinguer les homonymes.
:param teachers: Enseignants du devoir, triés pour garantir la stabilité.
:return: ID stable (12 caractères hexadécimaux).
:rtype: str
"""
payload = f"{due_on.isoformat()}|{normalized_text}".encode()
payload = (
f"{due_on.isoformat()}|{subject.casefold()}|"
f"{','.join(sorted(teacher.casefold() for teacher in teachers))}|{normalized_text}".encode()
)
return hashlib.sha1(payload, usedforsecurity=False).hexdigest()[:12]
@@ -313,7 +384,7 @@ def collect_homeworks(lessons: list[Lesson], target_date: date) -> list[Homework
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
se fait par texte, matière et enseignants normalisés (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.
@@ -321,15 +392,22 @@ def collect_homeworks(lessons: list[Lesson], target_date: date) -> list[Homework
:return: Liste unique de devoirs, triée par matière puis texte.
:rtype: list[Homework]
"""
by_text: dict[str, Homework] = {}
by_context: dict[tuple[str, tuple[str, ...], 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),
normalized_text = normalize_homework_text(block.text)
key = (
lesson.subject.casefold(),
tuple(sorted(teacher.casefold() for teacher in lesson.teachers)),
normalized_text,
)
if key not in by_context:
by_context[key] = Homework(
id=generate_homework_id(
target_date, normalized_text, lesson.subject, lesson.teachers
),
subject=lesson.subject,
teachers=lesson.teachers,
assigned_on=lesson.start.date(),
@@ -343,10 +421,17 @@ def collect_homeworks(lessons: list[Lesson], target_date: date) -> list[Homework
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),
normalized_text = normalize_homework_text(block.text)
key = (
lesson.subject.casefold(),
tuple(sorted(teacher.casefold() for teacher in lesson.teachers)),
normalized_text,
)
if key not in by_context:
by_context[key] = Homework(
id=generate_homework_id(
target_date, normalized_text, lesson.subject, lesson.teachers
),
subject=lesson.subject,
teachers=lesson.teachers,
assigned_on=block.date,
@@ -355,7 +440,7 @@ def collect_homeworks(lessons: list[Lesson], target_date: date) -> list[Homework
html=block.html,
)
return sorted(by_text.values(), key=lambda hw: (hw.subject.lower(), hw.text.lower()))
return sorted(by_context.values(), key=lambda hw: (hw.subject.lower(), hw.text.lower()))
def parse_ical(raw_ical: str) -> tuple[list[Lesson], list[Homework], list[SchoolEvent]]:
@@ -399,15 +484,28 @@ def parse_ical(raw_ical: str) -> tuple[list[Lesson], list[Homework], list[School
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):
normalized_categories = [_normalize_label(category) for category in categories]
holiday_kind: SchoolEventKind | None = None
if any(
category in ("conges", "vacances", "vacances scolaires")
for category in normalized_categories
):
holiday_kind = SchoolEventKind.HOLIDAY
elif any(
category in ("jour ferie", "jours feries", "ferie", "feries")
for category in normalized_categories
):
holiday_kind = SchoolEventKind.PUBLIC_HOLIDAY
# Événements de type vacances/congés/jours fériés (tout le jour).
if holiday_kind is not None:
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,
kind=holiday_kind,
label=label,
from_date=from_date,
to_date=to_date,
@@ -422,9 +520,15 @@ def parse_ical(raw_ical: str) -> tuple[list[Lesson], list[Homework], list[School
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:
normalized_status_categories = set(normalized_categories)
if status_value == "CANCELLED" or any(
"annul" in category for category in normalized_status_categories
):
status = LessonStatus.CANCELLED
elif "Cours - Cours déplacé" in categories:
elif any(
any(token in category for token in ("deplac", "changement de salle", "modifi"))
for category in normalized_status_categories
):
status = LessonStatus.MOVED
else:
status = LessonStatus.NORMAL
@@ -458,10 +562,11 @@ def parse_ical(raw_ical: str) -> tuple[list[Lesson], list[Homework], list[School
teachers=tuple(lesson_data["teachers"]),
rooms=tuple(lesson_data["rooms"]),
group=lesson_data["group"],
class_part=lesson_data["class_part"],
status=status,
content=content,
homework_blocks=homework_blocks,
)
)
return lessons, homeworks, school_events
return collapse_replaced_lessons(lessons), homeworks, school_events
+72
View File
@@ -0,0 +1,72 @@
"""Règles communes de statut et de consolidation des cours Pronote."""
from __future__ import annotations
import unicodedata
from collections import defaultdict
from ...models.agenda import Lesson, LessonStatus
def _normalize_status(value: str) -> str:
"""Normalise un statut Pronote pour une comparaison tolérante.
:param value: Statut brut fourni par Pronote.
:return: Statut minuscule sans accents.
:rtype: str
"""
decomposed = unicodedata.normalize("NFKD", value)
without_accents = "".join(char for char in decomposed if not unicodedata.combining(char))
return " ".join(without_accents.casefold().split())
def lesson_status(raw_status: object, canceled: bool = False) -> LessonStatus:
"""Convertit le statut Pronote en statut métier canonique.
L'indicateur d'annulation et les libellés annulé sont prioritaires sur un
éventuel libellé de déplacement. Les libellés inconnus restent normaux.
:param raw_status: Valeur du champ ``Lesson.status`` de pronotepy.
:param canceled: Indicateur ``Lesson.canceled`` de pronotepy.
:return: Statut métier canonique.
:rtype: LessonStatus
"""
status = _normalize_status(raw_status) if isinstance(raw_status, str) else ""
if canceled or "annul" in status:
return LessonStatus.CANCELLED
if any(token in status for token in ("deplac", "changement de salle", "modifi")):
return LessonStatus.MOVED
return LessonStatus.NORMAL
def collapse_replaced_lessons(lessons: list[Lesson]) -> list[Lesson]:
"""Supprime l'ancien exemplaire annulé lorsqu'un cours déplacé est présent.
Deux cours sont considérés comme les exemplaires d'une même modification
lorsqu'ils partagent les dates, la matière, les enseignants et le groupe,
et qu'une seule annulation et un seul déplacement se distinguent par leur
salle. Une annulation isolée ou ambiguë est toujours conservée.
:param lessons: Cours issus d'une source Pronote.
:return: Cours consolidés dans leur ordre d'origine.
:rtype: list[Lesson]
"""
grouped: dict[tuple[object, ...], list[tuple[int, Lesson]]] = defaultdict(list)
for index, lesson in enumerate(lessons):
key = (
lesson.start,
lesson.end,
lesson.subject.casefold(),
tuple(teacher.casefold() for teacher in lesson.teachers),
(lesson.group or "").casefold(),
)
grouped[key].append((index, lesson))
replaced_indexes: set[int] = set()
for siblings in grouped.values():
cancelled = [item for item in siblings if item[1].status == LessonStatus.CANCELLED]
moved = [item for item in siblings if item[1].status == LessonStatus.MOVED]
if len(cancelled) == 1 and len(moved) == 1 and cancelled[0][1].rooms != moved[0][1].rooms:
replaced_indexes.add(cancelled[0][0])
return [lesson for index, lesson in enumerate(lessons) if index not in replaced_indexes]
+4
View File
@@ -61,6 +61,10 @@ def lesson_to_vevent(lesson: Lesson) -> Event:
parts.append(f"Professeur(s): {', '.join(lesson.teachers)}")
if lesson.rooms:
parts.append(f"Salle(s): {', '.join(lesson.rooms)}")
if lesson.group:
parts.append(f"Groupe: {lesson.group}")
if lesson.class_part:
parts.append(f"Partie(s) de classe: {lesson.class_part}")
if lesson.content:
parts.append(f"Contenu: {lesson.content}")
event.add("description", "\n".join(parts))
+3 -1
View File
@@ -37,6 +37,7 @@ _URL_SECRET_RE = re.compile(
r"(?ix)[?&](?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)"
r"=([^&#\s]{3,})"
)
_URL_PLACEHOLDER_RE = re.compile(r"(?x)^(?:\{[^{}\r\n]+\}|<[^<>\r\n]+>|\.+|[•…*]+)$")
_EXTRA_NAMES = frozenset({"pronote_sync"})
@@ -185,7 +186,8 @@ def find_secrets(
)
if is_literal_secret:
findings.append(SecretFinding(relative_path, number, "affectation-litterale"))
if _URL_SECRET_RE.search(line):
url_match = _URL_SECRET_RE.search(line)
if url_match and not _URL_PLACEHOLDER_RE.fullmatch(url_match.group(1)):
findings.append(SecretFinding(relative_path, number, "parametre-url"))
return sorted(findings, key=lambda finding: (str(finding.path), finding.line, finding.rule))
+17
View File
@@ -11,6 +11,23 @@ from pronote_sync.errors import PipelineCriticalError, PipelineWarning
from pronote_sync.models.pronote import PronoteData
def test_help_documents_strict_dry_run_contract(capsys: pytest.CaptureFixture[str]) -> None:
"""L'aide CLI expose le contrat strict et l'incompatibilité QR/token.
:param capsys: Capture des sorties standard de pytest.
:return: None
"""
from pronote_sync.cli.main import main
with pytest.raises(SystemExit) as exc_info:
main(["--help"])
output = " ".join(capsys.readouterr().out.split())
assert exc_info.value.code == 0
assert "sans sortie distante ni état local persistant" in output
assert "PRONOTE_AUTH_MODE=qr_token" in output
def test_main_runs_composition_root_in_dry_run_with_requested_log_level(
mocker: MockerFixture,
) -> None:
+53
View File
@@ -0,0 +1,53 @@
BEGIN:VCALENDAR
VERSION:2.0
PRODID:-//pronote-sync tests//FR
X-WR-CALNAME:Fixture anonymisée
BEGIN:VEVENT
UID:variant-normal
DTSTART:20260910T080000Z
DTEND:20260910T090000Z
SUMMARY:Mathématiques
CATEGORIES:Cours
DESCRIPTION:<div>Matière : Mathématiques Professeur(s) : Enseignant A Salle(s) : 101 Groupe : Classe entière Partie(s) de classe : Groupe 1 <strong data-section="content">Contenu pédagogique :</strong><p>Leçon &amp; exemple</p><strong>Pour le 15/09/2026 :</strong><p>Lire le chapitre</p><strong>Donné le 10/09/2026 :</strong><p>Lire le chapitre</p></div>
END:VEVENT
BEGIN:VEVENT
UID:variant-duplicate
DTSTART:20260910T100000Z
DTEND:20260910T110000Z
SUMMARY:Mathématiques
CATEGORIES:Cours
DESCRIPTION:<div>Matière : Mathématiques Professeur(s) : Enseignant A Salle(s) : 102 <strong>Pour le 15/09/2026 :</strong><p>Lire le chapitre</p></div>
END:VEVENT
BEGIN:VEVENT
UID:variant-same-text-other-context
DTSTART:20260910T120000Z
DTEND:20260910T130000Z
SUMMARY:Sciences
CATEGORIES:Cours
DESCRIPTION:<div>Matière : Sciences Professeur(s) : Enseignant B Salle(s) : 103 <strong>Pour le 15/09/2026 :</strong><p>Lire le chapitre</p></div>
END:VEVENT
BEGIN:VEVENT
UID:variant-moved
DTSTART:20260911T080000Z
DTEND:20260911T090000Z
SUMMARY:Mathématiques
CATEGORIES:Cours - Cours modifié
DESCRIPTION:<div>Matière : Mathématiques Professeur : Enseignant A Salle : 204</div>
END:VEVENT
BEGIN:VEVENT
UID:variant-public-holiday
DTSTART;VALUE=DATE:20260914
DTEND;VALUE=DATE:20260915
SUMMARY:Journée fériée anonymisée
CATEGORIES:Jours fériés
DESCRIPTION:Journée sans cours.
END:VEVENT
BEGIN:VEVENT
UID:variant-holiday
DTSTART;VALUE=DATE:20261020
DTEND;VALUE=DATE:20261022
SUMMARY:Vacances anonymisées
CATEGORIES:Congés
DESCRIPTION:Période de vacances.
END:VEVENT
END:VCALENDAR
+19
View File
@@ -0,0 +1,19 @@
{
"version": "fixture-1",
"source": "ical",
"pronote_version": "2026-unknown",
"anonymized": true,
"scenarios": {
"normal_and_headers": "variant-normal",
"duplicate_homework_same_context": ["variant-normal", "variant-duplicate"],
"same_text_distinct_context": "variant-same-text-other-context",
"moved_lesson": "variant-moved",
"public_holiday": "variant-public-holiday",
"holiday_exclusive_end": "variant-holiday"
},
"invariants": [
"Aucune valeur d'authentification ou identité réelle n'est présente.",
"Les bornes de date d'un événement scolaire sont début inclusif et fin exclusive.",
"Les devoirs sont vérifiés après parsing dans les modèles métier finaux."
]
}
+92 -1
View File
@@ -8,7 +8,13 @@ from typing import Any, cast
import pytest
from pydantic import SecretStr
from pronote_sync.config.settings import AISettings, AppSettings, PronoteSettings, Settings
from pronote_sync.config.settings import (
AISettings,
AppSettings,
BlogSettings,
PronoteSettings,
Settings,
)
from pronote_sync.errors import PipelineCriticalError, PipelineWarning, PronoteAuthRotationError
from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent
from pronote_sync.models.blog import BlogArticle
@@ -422,6 +428,47 @@ def test_from_settings_password_mode_passes_auth_state_none(
assert constructed[0][1] is None
def test_from_settings_rejects_qr_token_dry_run_before_constructing_pronote_client(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""QR-token dry-run fails before authentication or data dependencies are created."""
import pronote_sync.pipeline.run as run_module
qr_pin_sentinel = "qr-pin-must-not-appear"
constructed: list[object] = []
class FailingClient:
"""Pronote client sentinel that makes unexpected construction explicit."""
def __init__(self, settings: PronoteSettings, *, auth_state: object) -> None:
"""Record and reject any unexpected client construction.
:param settings: Pronote settings supplied by the composition root.
:param auth_state: Authentication state supplied by the composition root.
"""
del settings, auth_state
constructed.append(object())
raise AssertionError("PronoteClient must not be constructed for QR-token dry-run")
monkeypatch.setattr(run_module, "PronoteClient", FailingClient)
with pytest.raises(ValueError) as exc_info:
PipelineRunner.from_settings(
Settings(
app=AppSettings(dry_run=True),
pronote=PronoteSettings(
auth_mode="qr_token",
qr_pin=SecretStr(qr_pin_sentinel),
),
)
)
assert "qr_token" in str(exc_info.value)
assert "dry-run" in str(exc_info.value)
assert qr_pin_sentinel not in str(exc_info.value)
assert constructed == []
def test_from_settings_qr_token_mode_passes_auth_state_instance(
monkeypatch: pytest.MonkeyPatch,
) -> None:
@@ -449,6 +496,50 @@ def test_from_settings_qr_token_mode_passes_auth_state_instance(
assert isinstance(constructed[0][1], PronoteAuthState)
@pytest.mark.parametrize("dry_run", [False, True])
def test_from_settings_configures_source_state_persistence_for_dry_run(
monkeypatch: pytest.MonkeyPatch,
dry_run: bool,
) -> None:
"""Composition disables source-state persistence only in dry-run mode."""
import pronote_sync.pipeline.run as run_module
blog_persistence: list[bool] = []
class RecordingBlogState:
"""Blog state factory recording its persistence configuration."""
def __init__(self, *, persistence_enabled: bool = True) -> None:
"""Record the requested persistence setting.
:param persistence_enabled: Whether disk writes are enabled.
"""
blog_persistence.append(persistence_enabled)
class RecordingClient:
"""Pronote client constructor accepting the injected auth state."""
def __init__(self, settings: PronoteSettings, *, auth_state: object) -> None:
"""Accept the composition-root dependencies.
:param settings: Pronote settings.
:param auth_state: Injected authentication state.
"""
del settings, auth_state
monkeypatch.setattr(run_module, "BlogRSSState", RecordingBlogState)
monkeypatch.setattr(run_module, "PronoteClient", RecordingClient)
PipelineRunner.from_settings(
Settings(
app=AppSettings(dry_run=dry_run),
blog=BlogSettings(enabled=True),
)
)
assert blog_persistence == [not dry_run]
def test_runner_reuses_ical_download_and_parse_within_one_run(
monkeypatch: pytest.MonkeyPatch,
pipeline_inputs: tuple[Lesson, Homework],
+1 -1
View File
@@ -67,7 +67,7 @@ class FakeClientXMPP:
:param handler: Fonction gestionnaire.
:raises: AssertionError si l'événement n'est pas supporté.
"""
if name not in ("session_start", "failed_auth", "disconnected"):
if name not in ("session_start", "failed_auth", "connection_failed", "disconnected"):
raise AssertionError(f"Unsupported event: {name}")
self.handlers.setdefault(name, []).append(handler)
+43
View File
@@ -35,6 +35,49 @@ def test_state_file_absent_empty_state(tmp_path: Path) -> None:
assert state.get_cache_headers() == (None, None)
def test_disabled_persistence_keeps_updates_in_memory_without_creating_file(tmp_path: Path) -> None:
"""Vérifie que la persistance désactivée conserve l'état uniquement en mémoire.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file, persistence_enabled=False)
state.add_guids(["guid-1"])
state.update_cache_headers("etag-123", "Wed, 01 Sep 2026 GMT")
assert state.get_known_guids() == frozenset({"guid-1"})
assert state.get_cache_headers() == ("etag-123", "Wed, 01 Sep 2026 GMT")
assert not state_file.exists()
def test_disabled_persistence_preserves_existing_file(tmp_path: Path) -> None:
"""Vérifie que la persistance désactivée ne modifie pas l'état déjà stocké.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
original_content = json.dumps(
{
"version": 1,
"known_guids": ["existing-guid"],
"etag": "old-etag",
"last_modified": "Tue, 31 Aug 2026 GMT",
}
)
state_file.write_text(original_content, encoding="utf-8")
state = BlogRSSState(state_file, persistence_enabled=False)
state.add_guids(["new-guid"])
state.update_cache_headers("new-etag", "Wed, 01 Sep 2026 GMT")
assert state.get_known_guids() == frozenset({"existing-guid", "new-guid"})
assert state.get_cache_headers() == ("new-etag", "Wed, 01 Sep 2026 GMT")
assert state_file.read_text(encoding="utf-8") == original_content
def test_add_guids_persists(tmp_path: Path) -> None:
"""Vérifie que l'ajout de GUID persiste dans le fichier JSON.
+27
View File
@@ -118,6 +118,33 @@ def test_main_detects_sensitive_url_parameter(
assert sentinel not in output
def test_main_ignores_documentation_url_placeholders(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
"""Ignore les marqueurs de remplacement utilisés dans une documentation.
:param secret_checker: Module du script sous test.
:param tmp_path: Répertoire temporaire représentant un dépôt.
:param capsys: Fixture de capture de sortie.
:return: None
"""
(tmp_path / "guide.md").write_text(
"\n".join(
(
"https://example.invalid/?icalsecurise={jeton}",
"https://example.invalid/?icalsecurise=••••••••",
"https://example.invalid/?icalsecurise=<token>",
"https://example.invalid/?icalsecurise=...",
)
)
+ "\n",
encoding="utf-8",
)
assert secret_checker.main([], root=tmp_path) == 0
assert "OK:" in capsys.readouterr().out
def test_staged_mode_inspects_only_paths_provided_by_git(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
+86
View File
@@ -0,0 +1,86 @@
"""Tests de sélection de la date cible du digest Pronote."""
from __future__ import annotations
from datetime import date, datetime
from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent, SchoolEventKind
from pronote_sync.pipeline.steps.fetch import resolve_target_date
def _lesson(day: date, status: LessonStatus = LessonStatus.NORMAL) -> Lesson:
"""Construit un cours minimal à une date donnée.
:param day: Date du cours.
:param status: Statut du cours.
:return: Cours de test.
:rtype: Lesson
"""
return Lesson(
id=day.isoformat(),
start=datetime.combine(day, datetime.min.time()),
end=datetime.combine(day, datetime.min.time().replace(hour=1)),
subject="Mathématiques",
group=None,
status=status,
content=None,
)
def _event(from_date: date, to_date: date) -> SchoolEvent:
"""Construit un événement scolaire à borne de fin exclusive.
:param from_date: Début inclusif.
:param to_date: Fin exclusive.
:return: Événement de vacances de test.
:rtype: SchoolEvent
"""
return SchoolEvent(
kind=SchoolEventKind.HOLIDAY,
label="Vacances anonymisées",
from_date=from_date,
to_date=to_date,
)
def test_target_date_prefers_tomorrow_effective_lesson() -> None:
"""Un cours effectif demain est prioritaire.
:return: None
"""
today = date(2026, 9, 11)
assert resolve_target_date(today, [_lesson(date(2026, 9, 12))], []) == date(2026, 9, 12)
def test_target_date_skips_empty_days_and_cancelled_lessons() -> None:
"""Le prochain cours effectif remplace un lendemain vide ou annulé.
:return: None
"""
today = date(2026, 9, 11)
lessons = [
_lesson(date(2026, 9, 12), LessonStatus.CANCELLED),
_lesson(date(2026, 9, 14)),
]
assert resolve_target_date(today, lessons, []) == date(2026, 9, 14)
def test_target_date_uses_exclusive_holiday_end_when_no_reprise_is_visible() -> None:
"""La fin exclusive d'une période couvrant demain devient la reprise connue.
:return: None
"""
today = date(2026, 10, 19)
holiday = _event(date(2026, 10, 20), date(2026, 11, 2))
assert resolve_target_date(today, [], [holiday]) == date(2026, 11, 2)
def test_target_date_falls_back_to_tomorrow_without_data() -> None:
"""Une liste vide valide sans événement conserve J+1 explicitement.
:return: None
"""
assert resolve_target_date(date(2026, 9, 11), [], []) == date(2026, 9, 12)
+148 -5
View File
@@ -294,8 +294,8 @@ def test_collect_homeworks_dedup() -> None:
id="lesson2",
start=datetime(2026, 9, 10, 10, 0),
end=datetime(2026, 9, 10, 11, 0),
subject="Physique",
teachers=("M. Martin",),
subject="Math",
teachers=("M. Dupont",),
rooms=("205",),
group=None,
status=LessonStatus.NORMAL,
@@ -343,8 +343,8 @@ def test_collect_homeworks_id_stability() -> None:
id="lesson2",
start=datetime(2026, 9, 10, 10, 0),
end=datetime(2026, 9, 10, 11, 0),
subject="Physique",
teachers=("M. Martin",),
subject="Math",
teachers=("M. Dupont",),
rooms=("205",),
group=None,
status=LessonStatus.NORMAL,
@@ -361,7 +361,150 @@ def test_collect_homeworks_id_stability() -> None:
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")
assert homeworks[0].id == generate_homework_id(
date(2026, 9, 10), "devoir commun", "Math", ("M. Dupont",)
)
def test_collect_homeworks_keeps_distinct_subjects() -> None:
"""Deux matières différentes conservent deux devoirs homonymes.
: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",),
group=None,
content=None,
homework_blocks=(
HomeworkBlock(
kind="due",
date=date(2026, 9, 10),
text="Devoir commun",
html="<p>Devoir commun</p>",
),
),
)
other_lesson = lesson.model_copy(
update={"id": "lesson2", "subject": "Physique", "teachers": ("M. Martin",)}
)
homeworks = collect_homeworks([lesson, other_lesson], target_date=date(2026, 9, 10))
assert len(homeworks) == 2
assert {homework.subject for homework in homeworks} == {"Math", "Physique"}
def test_collect_homeworks_deduplicates_teacher_order_variants() -> None:
"""Deux ordres d'enseignants équivalents ne créent pas de doublon.
:return: None
"""
homework_block = HomeworkBlock(
kind="due",
date=date(2026, 9, 10),
text="Devoir commun",
html="<p>Devoir commun</p>",
)
lesson = Lesson(
id="lesson1",
start=datetime(2026, 9, 10, 8, 0),
end=datetime(2026, 9, 10, 9, 0),
subject="Math",
teachers=("M. Martin", "M. Dupont"),
group=None,
content=None,
homework_blocks=(homework_block,),
)
other_lesson = lesson.model_copy(
update={"id": "lesson2", "teachers": ("M. Dupont", "M. Martin")}
)
homeworks = collect_homeworks([lesson, other_lesson], target_date=date(2026, 9, 10))
assert len(homeworks) == 1
def test_parse_body_accepts_html_variants_and_sanitizes() -> None:
"""Les attributs HTML, les espaces et le contenu actif sont traités correctement.
:return: None
"""
body = (
'<strong class="label"> Contenu pédagogique : </strong><p>Leçon &amp; exemple</p>'
'<strong data-kind="homework"> Pour le 10/09/2026 : </strong>'
'<p onclick="evil()">Lire <em>le chapitre</em></p><script>alert(1)</script>'
"<strong> Donné le 05/09/2026 : </strong><p>Noter &amp; relire</p>"
)
content, due_blocks, assigned_blocks = parse_body(body)
assert content == "Leçon & exemple"
assert due_blocks[0].date == date(2026, 9, 10)
assert due_blocks[0].text == "Lire le chapitre"
assert "<em>le chapitre</em>" in due_blocks[0].html
assert "onclick" not in due_blocks[0].html
assert "script" not in due_blocks[0].html.lower()
assert assigned_blocks[0].date == date(2026, 9, 5)
def test_parse_anonymized_variants_fixture() -> None:
"""Vérifie le corpus iCal anonymisé sur les statuts et le contexte des devoirs.
:return: None
"""
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-variants.ics"
lessons, _, school_events = parse_ical(fixture_path.read_text(encoding="utf-8"))
assert lessons[0].class_part == "Groupe 1"
assert lessons[0].homework_blocks[0].html == "<p>Lire le chapitre</p>"
assert any(lesson.status == LessonStatus.MOVED for lesson in lessons)
assert any(event.kind.value == "public_holiday" for event in school_events)
assert any(
event.from_date == date(2026, 10, 20) and event.to_date == date(2026, 10, 22)
for event in school_events
)
homeworks = collect_homeworks(lessons, target_date=date(2026, 9, 15))
assert len(homeworks) == 2
assert {homework.subject for homework in homeworks} == {"Mathématiques", "Sciences"}
def test_parse_ical_removes_old_replaced_lesson() -> None:
"""Un ancien cours annulé est retiré lorsqu'un cours déplacé le remplace.
:return: None
"""
raw_ical = """BEGIN:VCALENDAR
VERSION:2.0
BEGIN:VEVENT
UID:old
DTSTART:20260911T080000Z
DTEND:20260911T090000Z
SUMMARY:Mathématiques
CATEGORIES:Cours - Cours annulé
STATUS:CANCELLED
DESCRIPTION:<div>Matière : Mathématiques Professeur : Enseignant A Salle : 101</div>
END:VEVENT
BEGIN:VEVENT
UID:new
DTSTART:20260911T080000Z
DTEND:20260911T090000Z
SUMMARY:Mathématiques
CATEGORIES:Cours - Changement de salle
DESCRIPTION:<div>Matière : Mathématiques Professeur : Enseignant A Salle : 204</div>
END:VEVENT
END:VCALENDAR"""
lessons, _, _ = parse_ical(raw_ical)
assert len(lessons) == 1
assert lessons[0].status == LessonStatus.MOVED
assert lessons[0].rooms == ("204",)
def test_collect_homeworks_sorted() -> None:
+159
View File
@@ -17,10 +17,12 @@ from __future__ import annotations
import json
import logging
import os
from fcntl import LOCK_EX, LOCK_NB, LOCK_UN, flock
from pathlib import Path
import pytest
from pronote_sync.errors import PronoteAuthStateLockError
from pronote_sync.sources.pronote.auth_state import PronoteAuthState
@@ -35,6 +37,90 @@ def test_load_no_file_returns_none(tmp_path: Path) -> None:
assert state.load() is None
def test_disabled_persistence_keeps_credentials_in_memory_without_creating_file(
tmp_path: Path,
) -> None:
"""Vérifie que la persistance désactivée conserve les credentials en mémoire.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "auth.json"
credentials = {
"pronote_url": "https://example.com/pronote",
"username": "parent-1",
"password": "token-123", # pragma: allowlist secret
"uuid": "uuid-456",
}
state = PronoteAuthState(state_file, persistence_enabled=False)
state.save(credentials)
assert state.load() == credentials
assert not state_file.exists()
def test_disabled_persistence_preserves_existing_file(tmp_path: Path) -> None:
"""Vérifie que la persistance désactivée garde les nouveaux credentials en mémoire.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "auth.json"
original_credentials = {
"pronote_url": "https://example.com/pronote",
"username": "parent-1",
"password": "old-token", # pragma: allowlist secret
"uuid": "old-uuid",
}
original_content = json.dumps({"version": 1, "credentials": original_credentials}).encode()
new_credentials = {
"pronote_url": "https://example.com/pronote",
"username": "parent-1",
"password": "new-token", # pragma: allowlist secret
"uuid": "new-uuid",
}
state_file.write_bytes(original_content)
state = PronoteAuthState(state_file, persistence_enabled=False)
assert state.load() == original_credentials
state.save(new_credentials)
assert state.load() == new_credentials
assert state_file.read_bytes() == original_content
def test_disabled_persistence_clear_discards_in_memory_credentials_only(tmp_path: Path) -> None:
"""Vérifie que clear oublie l'état en mémoire sans modifier le fichier existant.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "auth.json"
original_credentials = {
"pronote_url": "https://example.com/pronote",
"username": "parent-1",
"password": "old-token", # pragma: allowlist secret
"uuid": "old-uuid",
}
original_content = json.dumps({"version": 1, "credentials": original_credentials}).encode()
credentials = {
"pronote_url": "https://example.com/pronote",
"username": "parent-1",
"password": "new-token", # pragma: allowlist secret
"uuid": "new-uuid",
}
state_file.write_bytes(original_content)
state = PronoteAuthState(state_file, persistence_enabled=False)
state.save(credentials)
state.clear()
assert state.load() is None
assert state_file.read_bytes() == original_content
def test_save_then_load_roundtrip(tmp_path: Path) -> None:
"""Vérifie que des credentials sauvegardés sont rechargés à l'identique.
@@ -206,3 +292,76 @@ def test_no_credentials_in_logs(tmp_path: Path, caplog: pytest.LogCaptureFixture
assert "SENTINEL_USER_ZZZ" not in caplog.text
assert "SENTINEL_PASSWORD_ZZZ" not in caplog.text
assert "SENTINEL_UUID_ZZZ" not in caplog.text
def test_lock_rejects_concurrent_access_with_a_redacted_dedicated_error(tmp_path: Path) -> None:
"""Vérifie qu'un verrou concurrent échoue immédiatement sans fuite interne.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / ".pronote_auth_state.json"
state = PronoteAuthState(state_file)
competing_state = PronoteAuthState(state_file)
with state.lock():
assert state_file.with_name(f"{state_file.name}.lock").exists()
with pytest.raises(PronoteAuthStateLockError) as exc_info:
with competing_state.lock():
pass
assert "BlockingIOError" not in str(exc_info.value)
assert exc_info.value.__cause__ is None
assert exc_info.value.__context__ is None
def test_lock_open_failure_does_not_log_sensitive_lock_path(
tmp_path: Path, caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Vérifie qu'un échec d'ouverture du verrou ne divulgue pas son chemin.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:param caplog: Fixture pytest pour capturer les logs.
:param monkeypatch: Fixture pytest pour remplacer l'ouverture du verrou.
:return: None
"""
sentinel_path = "/SENTINEL_LOCK_PATH_ZZZ/.pronote_auth_state.json.lock"
def raise_lock_open_error(*args: object, **kwargs: object) -> int:
"""Simule un refus d'ouverture portant un chemin sensible."""
del args, kwargs
raise OSError(13, "Permission denied", sentinel_path)
monkeypatch.setattr(os, "open", raise_lock_open_error)
with caplog.at_level(logging.ERROR):
with pytest.raises(PronoteAuthStateLockError) as exc_info:
with PronoteAuthState(tmp_path / ".pronote_auth_state.json").lock():
pass
assert "Impossible d'ouvrir le verrou d'état d'authentification Pronote" in caplog.text
assert "SENTINEL_LOCK_PATH_ZZZ" not in caplog.text
assert exc_info.value.__cause__ is None
assert exc_info.value.__context__ is None
def test_lock_is_released_when_the_protected_operation_raises(tmp_path: Path) -> None:
"""Vérifie que le verrou est libéré même si le bloc protégé échoue.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / ".pronote_auth_state.json"
lock_file = state_file.with_name(f"{state_file.name}.lock")
state = PronoteAuthState(state_file)
with pytest.raises(RuntimeError, match="échec simulé"):
with state.lock():
raise RuntimeError("échec simulé")
descriptor = os.open(lock_file, os.O_RDWR)
try:
flock(descriptor, LOCK_EX | LOCK_NB)
flock(descriptor, LOCK_UN)
finally:
os.close(descriptor)
+105
View File
@@ -9,6 +9,8 @@ from __future__ import annotations
import json
import logging
from collections.abc import Generator
from contextlib import contextmanager
from datetime import date, datetime
from pathlib import Path
from unittest.mock import MagicMock
@@ -231,6 +233,24 @@ def test_get_lessons_success(
assert lesson.content == "Lesson content"
def test_get_lessons_maps_moved_status(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
"""Mappe le statut ``Changement de salle`` de pronotepy vers ``MOVED``.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides.
:return: None
"""
mock_client = _make_lessons_mock_client(mocker)
mock_client.lessons.return_value[0].status = "Changement de salle"
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
lessons = PronoteClient(pronote_settings).get_lessons(date(2024, 9, 1), date(2024, 9, 30))
assert lessons[0].status == LessonStatus.MOVED
def test_get_homeworks_success(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None:
@@ -1203,6 +1223,91 @@ def test_no_raw_secrets_in_logs(
# --- Persistence of credentials after data operations ---
def test_qr_token_lock_covers_login_retrieval_and_credential_persistence(
mocker: pytest_mock.MockerFixture,
) -> None:
"""Vérifie que le verrou d'état couvre tout le cycle QR/token des cours.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
events: list[str] = []
credentials = {
"pronote_url": "https://pronote.example.com",
"username": "testuser",
"password": "persisted-token", # pragma: allowlist secret
"uuid": "persisted-uuid",
}
auth_state = mocker.MagicMock(spec=PronoteAuthState)
@contextmanager
def record_lock() -> Generator[None]:
events.append("lock_acquired")
try:
yield
finally:
events.append("lock_released")
def load_credentials() -> dict[str, str]:
"""Simule le chargement des credentials persistés."""
events.append("load")
return credentials
def get_no_lessons(*_: object) -> list[object]:
"""Simule une récupération de cours vide."""
events.append("lessons")
return []
def export_credentials() -> dict[str, str]:
"""Simule l'export des credentials courantes."""
events.append("export")
return credentials
def token_login(**_: object) -> MagicMock:
"""Simule le login par token."""
events.append("token_login")
return mock_client
auth_state.lock.side_effect = record_lock
auth_state.load.side_effect = load_credentials
auth_state.save.side_effect = lambda _: events.append("save")
mock_client = _make_lessons_mock_client(mocker)
mock_client.logged_in = True
mock_client.lessons.side_effect = get_no_lessons
mock_client.export_credentials.side_effect = export_credentials
mocker.patch(
"pronotepy.ParentClient.token_login",
side_effect=token_login,
)
settings = PronoteSettings(
url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent=None,
account_type="parent",
auth_mode="qr_token",
)
lessons = PronoteClient(settings, auth_state=auth_state).get_lessons(
date(2024, 9, 1), date(2024, 9, 30)
)
assert lessons == []
assert events == [
"lock_acquired",
"load",
"token_login",
"export",
"save",
"lessons",
"export",
"save",
"lock_released",
]
def _make_auth_state_mock(mocker: pytest_mock.MockerFixture) -> MagicMock:
"""Retourne un mock de PronoteAuthState sans credentials persistés.
+91
View File
@@ -0,0 +1,91 @@
"""Tests des règles communes de statut et de consolidation des cours."""
from __future__ import annotations
from datetime import datetime
import pytest
from pronote_sync.models.agenda import Lesson, LessonStatus
from pronote_sync.sources.pronote.lessons import collapse_replaced_lessons, lesson_status
def _lesson(status: LessonStatus, room: str) -> Lesson:
"""Construit un cours minimal pour les tests de consolidation.
:param status: Statut métier du cours.
:param room: Salle du cours.
:return: Cours de test.
:rtype: Lesson
"""
return Lesson(
id=f"lesson-{room}",
start=datetime(2026, 9, 11, 8, 0),
end=datetime(2026, 9, 11, 9, 0),
subject="Mathématiques",
teachers=("Enseignant A",),
rooms=(room,),
group="Classe entière",
status=status,
content=None,
)
@pytest.mark.parametrize(
("raw_status", "canceled", "expected"),
[
("Changement de salle", False, LessonStatus.MOVED),
("Cours déplacé", False, LessonStatus.MOVED),
("Cours annulé", False, LessonStatus.CANCELLED),
(None, True, LessonStatus.CANCELLED),
("Statut inconnu", False, LessonStatus.NORMAL),
],
)
def test_lesson_status_maps_pronotepy_variants(
raw_status: object, canceled: bool, expected: LessonStatus
) -> None:
"""Convertit les statuts Pronote connus sans interpréter les inconnus.
:param raw_status: Statut brut simulé.
:param canceled: Indicateur d'annulation simulé.
:param expected: Statut métier attendu.
:return: None
"""
assert lesson_status(raw_status, canceled) == expected
def test_collapse_replaced_lessons_keeps_effective_lesson() -> None:
"""Retire l'ancien exemplaire annulé d'un changement de salle.
:return: None
"""
cancelled = _lesson(LessonStatus.CANCELLED, "101")
moved = _lesson(LessonStatus.MOVED, "204")
result = collapse_replaced_lessons([cancelled, moved])
assert result == [moved]
def test_collapse_replaced_lessons_keeps_real_cancellation() -> None:
"""Conserve une annulation lorsqu'aucun cours déplacé ne la remplace.
:return: None
"""
cancelled = _lesson(LessonStatus.CANCELLED, "101")
assert collapse_replaced_lessons([cancelled]) == [cancelled]
def test_collapse_replaced_lessons_keeps_ambiguous_cancellations() -> None:
"""Conserve les annulations lorsqu'une paire de remplacement est ambiguë.
:return: None
"""
cancelled_old = _lesson(LessonStatus.CANCELLED, "101")
cancelled_real = _lesson(LessonStatus.CANCELLED, "305")
moved = _lesson(LessonStatus.MOVED, "204")
result = collapse_replaced_lessons([cancelled_old, cancelled_real, moved])
assert result == [cancelled_old, cancelled_real, moved]
+686 -8
View File
@@ -17,7 +17,7 @@ from unittest.mock import patch
import pytest
from pydantic import SecretStr
from pronote_sync.channels.xmpp import XmppChannel, XmppMessage
from pronote_sync.channels.xmpp import SyncXmppChannel, XmppChannel, XmppMessage
from pronote_sync.config.settings import XmppSettings
from pronote_sync.models.agenda import Lesson, TheoreticalLesson
from pronote_sync.models.blog import BlogArticle, ExternalInfo
@@ -48,11 +48,15 @@ class FakeClientXMPP:
self._should_disconnect_early = False
self._host_used: str | None = None
self._port_used: int | None = None
# New modes for realistic failure simulation
self._connect_mode: str = "ok" # "ok", "pending", "connection_failed"
self._disconnect_mode: str = "ok" # "ok", "pending"
self._fire_connection_failed: bool = False
def add_event_handler(
self, name: str, pointer: Callable[..., object], disposable: bool = False
) -> None:
if name not in ("session_start", "failed_auth", "disconnected"):
if name not in ("session_start", "failed_auth", "connection_failed", "disconnected"):
raise AssertionError(f"Unsupported event: {name}")
self.handlers.setdefault(name, []).append(pointer)
@@ -63,10 +67,24 @@ class FakeClientXMPP:
self.connected = True
self._host_used = host
self._port_used = port
# Schedule event handlers to fire after connect returns
loop.call_soon(self._fire_events)
future.set_result(True)
return future
if self._connect_mode == "pending":
# Never resolves - simulates connection timeout
return future
elif self._connect_mode == "connection_failed":
# Resolves connect future but fires connection_failed event
loop.call_soon(self._fire_connection_failed_event)
future.set_result(True)
return future
else:
# Schedule event handlers to fire after connect returns
loop.call_soon(self._fire_events)
future.set_result(True)
return future
def _fire_connection_failed_event(self) -> None:
"""Fire connection_failed event for testing."""
self._fire("connection_failed")
def _fire_events(self) -> None:
if self._should_disconnect_early:
@@ -86,8 +104,13 @@ class FakeClientXMPP:
loop = asyncio.get_event_loop()
future: asyncio.Future[bool] = loop.create_future()
self.disconnected = True
future.set_result(True)
return future
if self._disconnect_mode == "pending":
# Never resolves - simulates cleanup timeout
return future
else:
future.set_result(True)
return future
def send_message(
self, mto: object, mbody: str | None = None, mtype: str | None = None, **kwargs: object
@@ -681,6 +704,339 @@ class TestXmppChannelSend:
jid_arg = call_args.args[0]
assert jid_arg == "bot@example.com/myresource"
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
@pytest.mark.asyncio
async def test_connect_timeout_returns_false(self, caplog: pytest.LogCaptureFixture) -> None:
"""Test que connect_timeout retourne False quand connect() ne résout pas.
:param caplog: Fixture pytest pour capturer les logs.
"""
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
tls_mode="starttls",
connect_timeout=0.05,
timeout=30,
cleanup_timeout=0.01,
)
class PendingConnectClient(FakeClientXMPP):
def __init__(self, jid: str, password: str) -> None:
super().__init__(jid, password)
self._connect_mode = "pending"
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=PendingConnectClient):
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
result = await channel.send_async(msg)
assert result is False
# Vérifier que le log contient "connexion"
logs = caplog.text
assert "connexion" in logs.lower()
@pytest.mark.asyncio
async def test_cancelled_error_cancels_pending_tasks(self) -> None:
"""Test que l'annulation de send_async annule les tâches encore en attente.
Un ``connect()`` qui ne résout jamais et une annulation de la tâche
appelante doivent entraîner la cancellation de la future de connexion
(idempotence de ``_cancel_pending`` sur les chemins d'exception).
"""
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
tls_mode="starttls",
connect_timeout=15,
timeout=30,
cleanup_timeout=0.01,
)
class PendingConnectClient(FakeClientXMPP):
"""Client dont ``connect()`` retourne une future jamais résolue."""
def __init__(self, jid: str, password: str) -> None:
super().__init__(jid, password)
self._connect_mode = "pending"
self.connect_future: asyncio.Future[bool] | None = None
def connect(
self, host: str | None = None, port: int | None = None
) -> asyncio.Future[bool]:
future = super().connect(host, port)
self.connect_future = future
return future
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=PendingConnectClient):
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
client = PendingConnectClient("bot@example.com", "secret123")
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=lambda j, p: client):
task = asyncio.ensure_future(channel.send_async(msg))
# Laisse asyncio.wait démarrer et la future de connexion rester en attente
await asyncio.sleep(0.05)
task.cancel()
result = await task
assert result is False
assert client.connect_future is not None
assert client.connect_future.cancelled()
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
@pytest.mark.asyncio
async def test_connection_failed_event_returns_false(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""Test que connection_failed event retourne False rapidement.
:param caplog: Fixture pytest pour capturer les logs.
"""
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
tls_mode="starttls",
connect_timeout=15,
timeout=30,
cleanup_timeout=0.01,
)
class ConnectionFailedClient(FakeClientXMPP):
def __init__(self, jid: str, password: str) -> None:
super().__init__(jid, password)
self._connect_mode = "connection_failed"
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ConnectionFailedClient):
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
result = await channel.send_async(msg)
assert result is False
# Vérifier que le log contient une mention d'échec réseau
logs = caplog.text
assert "réseau" in logs.lower() or "connexion" in logs.lower()
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
@pytest.mark.asyncio
async def test_cleanup_timeout_does_not_hang(self, caplog: pytest.LogCaptureFixture) -> None:
"""Test que cleanup_timeout ne bloque pas quand disconnect() ne résout pas.
:param caplog: Fixture pytest pour capturer les logs.
"""
import logging
caplog.set_level(logging.DEBUG)
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
tls_mode="starttls",
connect_timeout=0.05,
timeout=30,
cleanup_timeout=0.01,
)
class HangingDisconnectClient(FakeClientXMPP):
def __init__(self, jid: str, password: str) -> None:
super().__init__(jid, password)
self._disconnect_mode = "pending"
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=HangingDisconnectClient):
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
# Should complete quickly despite hanging disconnect
result = await asyncio.wait_for(channel.send_async(msg), timeout=0.5)
assert result is True
# Vérifier que le log contient une mention de timeout de nettoyage
logs = caplog.text
assert "déconnexion" in logs.lower()
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
@pytest.mark.asyncio
async def test_tls_mode_direct_config(self) -> None:
"""Test que tls_mode='direct' configure enable_direct_tls=True et enable_starttls=False.
:return: None
"""
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
tls_mode="direct",
timeout=30,
)
class InspectClient(FakeClientXMPP):
def __init__(self, jid: str, password: str) -> None:
super().__init__(jid, password)
with patch("pronote_sync.channels.xmpp.ClientXMPP") as mock_cls:
mock_cls.return_value = InspectClient("bot@example.com", "secret123")
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
await channel.send_async(msg)
client_instance = mock_cls.return_value
assert client_instance.enable_direct_tls is True
assert client_instance.enable_starttls is False
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
@pytest.mark.asyncio
async def test_tls_mode_starttls_config(self) -> None:
"""Test que tls_mode='starttls' configure enable_starttls=True et enable_direct_tls=False.
:return: None
"""
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
tls_mode="starttls",
timeout=30,
)
class InspectClient(FakeClientXMPP):
def __init__(self, jid: str, password: str) -> None:
super().__init__(jid, password)
with patch("pronote_sync.channels.xmpp.ClientXMPP") as mock_cls:
mock_cls.return_value = InspectClient("bot@example.com", "secret123")
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
await channel.send_async(msg)
client_instance = mock_cls.return_value
assert client_instance.enable_starttls is True
assert client_instance.enable_direct_tls is False
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
@pytest.mark.asyncio
async def test_tls_mode_disabled_config(self) -> None:
"""Test que tls_mode='disabled' avec host='127.0.0.1' configure TLS désactivé.
:return: None
"""
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"),
host="127.0.0.1",
port=5222,
to="parent@example.com",
resource="pronote-sync",
tls_mode="disabled",
timeout=30,
)
class InspectClient(FakeClientXMPP):
def __init__(self, jid: str, password: str) -> None:
super().__init__(jid, password)
with patch("pronote_sync.channels.xmpp.ClientXMPP") as mock_cls:
mock_cls.return_value = InspectClient("bot@example.com", "secret123")
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
await channel.send_async(msg)
client_instance = mock_cls.return_value
assert client_instance.enable_direct_tls is False
assert client_instance.enable_starttls is False
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
@pytest.mark.asyncio
async def test_no_secret_leak_in_connection_failure_log(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""Test qu'aucun secret ne fuit dans les logs en cas d'échec de connexion.
:param caplog: Fixture pytest pour capturer les logs.
"""
sentinel_password = "SECRET_PASSWORD_XMPP_12345" # pragma: allowlist secret
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr(sentinel_password),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
tls_mode="starttls",
timeout=30,
)
class ConnectionFailedClient(FakeClientXMPP):
def __init__(self, jid: str, password: str) -> None:
super().__init__(jid, password)
self._connect_mode = "connection_failed"
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ConnectionFailedClient):
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
await channel.send_async(msg)
# Vérifier que le mot de passe sentinelle n'apparaît pas dans les logs
logs = caplog.text
assert sentinel_password not in logs
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
@pytest.mark.asyncio
async def test_no_secret_leak_in_connect_timeout_log(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""Test qu'aucun secret ne fuit dans les logs en cas de timeout de connexion.
:param caplog: Fixture pytest pour capturer les logs.
"""
sentinel_password = "SECRET_PASSWORD_XMPP_12345"
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr(sentinel_password),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
tls_mode="starttls",
connect_timeout=0.05,
timeout=30,
cleanup_timeout=0.01,
)
class PendingConnectClient(FakeClientXMPP):
def __init__(self, jid: str, password: str) -> None:
super().__init__(jid, password)
self._connect_mode = "pending"
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=PendingConnectClient):
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
await channel.send_async(msg)
# Vérifier que le mot de passe sentinelle n'apparaît pas dans les logs
logs = caplog.text
assert sentinel_password not in logs
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
@pytest.mark.asyncio
async def test_send_async_tls_direct_config(self) -> None:
@@ -816,6 +1172,328 @@ class TestXmppChannelSend:
result = await channel.send_async(msg)
assert result is True
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
@pytest.mark.asyncio
async def test_send_async_connect_raises_exception_returns_false(self) -> None:
"""Test que connect() levant une exception retourne False.
:return: None
"""
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
tls_mode="starttls",
timeout=30,
)
class ConnectExceptionClient(FakeClientXMPP):
def connect(
self, host: str | None = None, port: int | None = None
) -> asyncio.Future[bool]:
raise ConnectionError("Network unreachable")
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ConnectExceptionClient):
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
result = await channel.send_async(msg)
assert result is False
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
@pytest.mark.asyncio
async def test_send_async_connect_future_error_returns_false(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""Test que la future de connect() résolue avec une exception retourne False.
Le message d'erreur doit être expurgé et ne pas contenir les secrets.
:param caplog: Fixture pytest pour capturer les logs.
"""
sentinel_password = "SECRET_PASSWORD_XMPP_CONNECT_FUT" # pragma: allowlist secret
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr(sentinel_password),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
tls_mode="starttls",
timeout=30,
)
class ConnectFutureErrorClient(FakeClientXMPP):
def connect(
self, host: str | None = None, port: int | None = None
) -> asyncio.Future[bool]:
loop = asyncio.get_event_loop()
future: asyncio.Future[bool] = loop.create_future()
future.set_exception(ConnectionError("Network unreachable"))
return future
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ConnectFutureErrorClient):
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
result = await channel.send_async(msg)
assert result is False
assert "Échec de connexion XMPP" in caplog.text
assert sentinel_password not in caplog.text
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
@pytest.mark.asyncio
async def test_send_async_session_failure_after_connect_returns_false(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""Test qu'un échec de session après connexion résolue retourne False.
Le connect future est résolu avec succès, puis l'événement
``failed_auth`` arrive pendant l'attente de session.
:param caplog: Fixture pytest pour capturer les logs.
"""
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
tls_mode="starttls",
timeout=5,
cleanup_timeout=0.01,
)
class LateAuthFailClient(FakeClientXMPP):
def connect(
self, host: str | None = None, port: int | None = None
) -> asyncio.Future[bool]:
loop = asyncio.get_event_loop()
future: asyncio.Future[bool] = loop.create_future()
self.connected = True
self._host_used = host
self._port_used = port
loop.call_later(0.01, lambda: self._fire("failed_auth"))
future.set_result(True)
return future
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=LateAuthFailClient):
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
result = await channel.send_async(msg)
assert result is False
assert "authentification" in caplog.text.lower()
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
@pytest.mark.asyncio
async def test_send_async_cancelled_returns_false(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""Test que l'annulation de la tâche retourne False sans lever.
Le contrat du canal impose un retour booléen : ``CancelledError`` doit
être interceptée et convertie en ``False``.
:param caplog: Fixture pytest pour capturer les logs.
"""
import logging
caplog.set_level(logging.DEBUG)
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
tls_mode="starttls",
connect_timeout=15,
timeout=30,
cleanup_timeout=0.01,
)
class PendingConnectClient(FakeClientXMPP):
def __init__(self, jid: str, password: str) -> None:
super().__init__(jid, password)
self._connect_mode = "pending"
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=PendingConnectClient):
channel = XmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
task = asyncio.ensure_future(channel.send_async(msg))
await asyncio.sleep(0.001)
task.cancel()
result = await asyncio.wait_for(task, timeout=1.0)
assert result is False
assert "annulé" in caplog.text.lower()
class TestSecretValues:
"""Tests pour la fonction _secret_values."""
def test_secret_values_with_all_secrets(self) -> None:
"""Test que _secret_values retourne tous les secrets.
:return: None
"""
from pronote_sync.channels.xmpp import _secret_values
settings = XmppSettings(
jid="bot@example.com",
password=SecretStr("secret123"),
to="parent@example.com",
)
secrets = _secret_values(settings)
assert len(secrets) == 3
assert "bot@example.com" in secrets
assert settings.password in secrets
assert "parent@example.com" in secrets
def test_secret_values_with_none_values(self) -> None:
"""Test que _secret_values filtre les valeurs None.
:return: None
"""
from pronote_sync.channels.xmpp import _secret_values
settings = XmppSettings(
jid=None,
password=None,
to=None,
)
secrets = _secret_values(settings)
assert len(secrets) == 0
def test_secret_values_with_some_none(self) -> None:
"""Test que _secret_values gère les valeurs partiellement None.
:return: None
"""
from pronote_sync.channels.xmpp import _secret_values
settings = XmppSettings(
jid="bot@example.com",
password=None,
to="parent@example.com",
)
secrets = _secret_values(settings)
assert len(secrets) == 2
assert "bot@example.com" in secrets
assert "parent@example.com" in secrets
class TestSyncXmppChannel:
"""Tests unitaires pour la classe SyncXmppChannel."""
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
def test_sync_send_dry_run_returns_true(self) -> None:
"""Test que SyncXmppChannel en dry_run retourne True.
:return: None
"""
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
use_tls=True,
timeout=30,
)
channel = SyncXmppChannel(settings, dry_run=True)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
result = channel.send(msg)
assert result is True
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
def test_sync_send_success_returns_true(self) -> None:
"""Test que SyncXmppChannel.send retourne True en cas de succès.
:return: None
"""
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
use_tls=True,
timeout=30,
)
channel = SyncXmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
result = channel.send(msg)
assert result is True
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
def test_sync_send_exception_returns_false(self) -> None:
"""Test que SyncXmppChannel.send retourne False en cas d'exception.
:return: None
"""
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
use_tls=True,
timeout=30,
)
class ErrorClient(FakeClientXMPP):
def connect(
self, host: str | None = None, port: int | None = None
) -> asyncio.Future[bool]:
raise RuntimeError("Connection failed")
with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ErrorClient):
channel = SyncXmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
result = channel.send(msg)
assert result is False
@patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP)
def test_sync_send_asyncio_run_error_returns_false(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""Test que send retourne False si asyncio.run lève une exception.
:param caplog: Fixture pytest pour capturer les logs.
"""
settings = XmppSettings(
enabled=True,
jid="bot@example.com",
password=SecretStr("secret123"),
host="xmpp.example.com",
port=5222,
to="parent@example.com",
resource="pronote-sync",
use_tls=True,
timeout=30,
)
channel = SyncXmppChannel(settings, dry_run=False)
msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None)
with patch(
"pronote_sync.channels.xmpp.asyncio.run",
side_effect=RuntimeError("Boucle événementielle indisponible"),
):
result = channel.send(msg)
assert result is False
assert "erreur" in caplog.text.lower()
class TestXmppChannelSecurity:
"""Tests de sécurité pour XmppChannel (non-fuite de secrets).
+116 -14
View File
@@ -99,19 +99,52 @@ class TestTimeoutConstraints:
class TestTlsPolicy:
"""Tests de la politique TLS pour le champ use_tls."""
def test_use_tls_false_with_remote_host_rejected(self) -> None:
"""Vérifie que use_tls=False avec un hôte distant est rejeté.
def test_tls_disabled_with_remote_host_rejected(self) -> None:
"""Vérifie que tls_mode="disabled" avec un hôte distant est rejeté.
:raises ValidationError: Si use_tls=False et host n'est pas une boucle locale.
:raises ValidationError: Si tls_mode="disabled" et host n'est pas une boucle locale.
"""
with pytest.raises(ValidationError) as exc_info:
XmppSettings(use_tls=False, host="talk.example.com")
assert "use_tls" in str(exc_info.value).lower() or "tls" in str(exc_info.value).lower()
XmppSettings(tls_mode="disabled", host="talk.example.com")
assert "tls" in str(exc_info.value).lower()
def test_use_tls_false_with_remote_host_rejected_when_enabled(self) -> None:
def test_tls_disabled_with_remote_host_rejected_when_enabled(self) -> None:
"""Vérifie que la politique TLS s'applique même quand le canal est activé."""
with pytest.raises(ValidationError):
XmppSettings(enabled=True, use_tls=False, host="talk.example.com")
XmppSettings(enabled=True, tls_mode="disabled", host="talk.example.com")
def test_use_tls_false_maps_to_starttls(self) -> None:
"""Vérifie que l'alias obsolète use_tls=False mappe vers tls_mode='starttls'.
:return: Vérifie le mapping et l'émission d'un DeprecationWarning.
:rtype: None
"""
with pytest.warns(DeprecationWarning):
settings = XmppSettings(use_tls=False, host="talk.example.com")
assert settings.tls_mode == "starttls"
assert settings.host == "talk.example.com"
def test_use_tls_true_maps_to_direct(self) -> None:
"""Vérifie que l'alias obsolète use_tls=True mappe vers tls_mode='direct'.
:return: Vérifie le mapping et l'émission d'un DeprecationWarning.
:rtype: None
"""
with pytest.warns(DeprecationWarning):
settings = XmppSettings(use_tls=True, host="talk.example.com")
assert settings.tls_mode == "direct"
assert settings.host == "talk.example.com"
def test_default_tls_mode_is_starttls(self) -> None:
"""Vérifie que tls_mode vaut 'starttls' par défaut avec les timeouts par défaut.
:return: Vérifie tls_mode='starttls', connect_timeout=15, cleanup_timeout=10.
:rtype: None
"""
settings = XmppSettings()
assert settings.tls_mode == "starttls"
assert settings.connect_timeout == 15
assert settings.cleanup_timeout == 10
def test_use_tls_false_with_localhost_allowed(self) -> None:
"""Vérifie que use_tls=False avec localhost est autorisé.
@@ -120,7 +153,7 @@ class TestTlsPolicy:
:rtype: None
"""
settings = XmppSettings(use_tls=False, host="localhost")
assert settings.use_tls is False
assert settings.tls_mode == "starttls"
assert settings.host == "localhost"
def test_use_tls_false_with_127_allowed(self) -> None:
@@ -130,7 +163,7 @@ class TestTlsPolicy:
:rtype: None
"""
settings = XmppSettings(use_tls=False, host="127.0.0.1")
assert settings.use_tls is False
assert settings.tls_mode == "starttls"
assert settings.host == "127.0.0.1"
def test_use_tls_false_with_ipv6_loopback_allowed(self) -> None:
@@ -140,7 +173,7 @@ class TestTlsPolicy:
:rtype: None
"""
settings = XmppSettings(use_tls=False, host="::1")
assert settings.use_tls is False
assert settings.tls_mode == "starttls"
assert settings.host == "::1"
def test_use_tls_true_with_remote_host_allowed(self) -> None:
@@ -150,7 +183,7 @@ class TestTlsPolicy:
:rtype: None
"""
settings = XmppSettings(use_tls=True, host="talk.example.com")
assert settings.use_tls is True
assert settings.tls_mode == "direct"
assert settings.host == "talk.example.com"
def test_use_tls_true_with_empty_host_allowed(self) -> None:
@@ -160,7 +193,7 @@ class TestTlsPolicy:
:rtype: None
"""
settings = XmppSettings(use_tls=True, host="")
assert settings.use_tls is True
assert settings.tls_mode == "direct"
assert settings.host == ""
@@ -173,7 +206,7 @@ class TestNoSecretInErrorMessages:
Crée une instance avec des valeurs sensibles et vérifie que l'erreur de validation
ne contient pas ces valeurs dans son message.
:raises ValidationError: Si use_tls=False avec un hôte non-local.
:raises ValidationError: Si tls_mode="disabled" avec un hôte non-local.
:return: Vérifie que le message d'erreur ne contient pas les secrets.
:rtype: None
"""
@@ -184,7 +217,7 @@ class TestNoSecretInErrorMessages:
with pytest.raises(ValidationError) as exc_info:
XmppSettings(
use_tls=False,
tls_mode="disabled",
host="talk.example.com",
jid=sentinel_jid,
password=sentinel_password,
@@ -197,3 +230,72 @@ class TestNoSecretInErrorMessages:
assert "test_password_123" not in error_message
assert "test_to@example.com" not in error_message
assert "secret" not in error_message
class TestXmppSettingsTlsMode:
"""Tests supplémentaires pour les validateurs de XmppSettings."""
def test_use_tls_true_maps_to_direct_with_warning(self) -> None:
"""Vérifie que use_tls=True mappe vers tls_mode='direct' avec DeprecationWarning.
:return: Vérifie le mapping et l'émission d'un DeprecationWarning.
:rtype: None
"""
with pytest.warns(DeprecationWarning):
settings = XmppSettings(use_tls=True, host="talk.example.com")
assert settings.tls_mode == "direct"
def test_use_tls_false_maps_to_starttls_with_warning(self) -> None:
"""Vérifie que use_tls=False mappe vers tls_mode='starttls' avec DeprecationWarning.
:return: Vérifie le mapping et l'émission d'un DeprecationWarning.
:rtype: None
"""
with pytest.warns(DeprecationWarning):
settings = XmppSettings(use_tls=False, host="talk.example.com")
assert settings.tls_mode == "starttls"
def test_tls_mode_disabled_rejected_for_remote(self) -> None:
"""Vérifie que tls_mode='disabled' avec un hôte distant est rejeté.
:raises ValidationError: Si tls_mode="disabled" et host n'est pas une boucle locale.
"""
with pytest.raises(ValidationError) as exc_info:
XmppSettings(tls_mode="disabled", host="example.com")
assert "tls" in str(exc_info.value).lower()
def test_tls_mode_disabled_allowed_for_loopback(self) -> None:
"""Vérifie que tls_mode='disabled' avec host='127.0.0.1' est valide.
:return: Vérifie que XmppSettings(tls_mode='disabled', host='127.0.0.1') est valide.
:rtype: None
"""
settings = XmppSettings(tls_mode="disabled", host="127.0.0.1")
assert settings.tls_mode == "disabled"
assert settings.host == "127.0.0.1"
def test_connect_timeout_must_be_positive(self) -> None:
"""Vérifie que connect_timeout doit être strictement positif.
:raises ValidationError: Si connect_timeout <= 0.
"""
with pytest.raises(ValidationError) as exc_info:
XmppSettings(connect_timeout=0)
assert "connect_timeout" in str(exc_info.value).lower()
with pytest.raises(ValidationError) as exc_info:
XmppSettings(connect_timeout=-1)
assert "connect_timeout" in str(exc_info.value).lower()
def test_cleanup_timeout_must_be_positive(self) -> None:
"""Vérifie que cleanup_timeout doit être strictement positif.
:raises ValidationError: Si cleanup_timeout <= 0.
"""
with pytest.raises(ValidationError) as exc_info:
XmppSettings(cleanup_timeout=0)
assert "cleanup_timeout" in str(exc_info.value).lower()
with pytest.raises(ValidationError) as exc_info:
XmppSettings(cleanup_timeout=-1)
assert "cleanup_timeout" in str(exc_info.value).lower()