Compare commits
8 Commits
fix/m1-sca
...
fix/m4-fix
| Author | SHA1 | Date | |
|---|---|---|---|
|
bb1f90bf5f
|
|||
|
9d9a55ed40
|
|||
|
1b550aa818
|
|||
|
e0cd073f9a
|
|||
|
f2be495f55
|
|||
|
f3177e70f6
|
|||
|
cb621c15f4
|
|||
|
f82e79360b
|
@@ -1,5 +1,7 @@
|
|||||||
# --- Pronote ---
|
# --- Pronote ---
|
||||||
PRONOTE_ICAL_URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024
|
PRONOTE_ICAL_URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024
|
||||||
|
PRONOTE_URL=https://college.ent/pronote/eleve.html
|
||||||
|
PRONOTE_ACCOUNT_TYPE=parent
|
||||||
PRONOTE_USERNAME=parent.dupont
|
PRONOTE_USERNAME=parent.dupont
|
||||||
PRONOTE_PASSWORD=your_secure_password
|
PRONOTE_PASSWORD=your_secure_password
|
||||||
PRONOTE_ENT=monbureaunumerique
|
PRONOTE_ENT=monbureaunumerique
|
||||||
|
|||||||
@@ -26,7 +26,7 @@ repos:
|
|||||||
name: mypy
|
name: mypy
|
||||||
entry: mypy
|
entry: mypy
|
||||||
language: python
|
language: python
|
||||||
additional_dependencies: ["mypy>=1.10.0", "pydantic>=2.0.0", "pydantic-settings>=2.0.0", "pytest>=8.0.0"]
|
additional_dependencies: ["mypy>=1.10.0", "pydantic>=2.0.0", "pydantic-settings>=2.0.0", "pytest>=8.0.0", "types-requests>=2.31.0", "icalendar>=5.0.0", "pronotepy>=2.15.0", "responses>=0.25.0", "pytest-mock>=3.10.0"]
|
||||||
types: [python]
|
types: [python]
|
||||||
pass_filenames: true
|
pass_filenames: true
|
||||||
|
|
||||||
|
|||||||
@@ -139,11 +139,11 @@
|
|||||||
"type": "Hex High Entropy String",
|
"type": "Hex High Entropy String",
|
||||||
"filename": "GUIDE_DEV_PYTHON.md",
|
"filename": "GUIDE_DEV_PYTHON.md",
|
||||||
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
|
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
|
||||||
"is_secret": false,
|
|
||||||
"is_verified": true,
|
"is_verified": true,
|
||||||
"line_number": 5117
|
"line_number": 5124,
|
||||||
|
"is_secret": false
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
},
|
},
|
||||||
"generated_at": "2026-09-05T21:51:55Z"
|
"generated_at": "2026-09-06T13:36:29Z"
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -285,6 +285,8 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati
|
|||||||
| `PRONOTE_AGENDA_SOURCE` | Source pour l'agenda (`auto`, `ical`, `pronotepy`). | `auto` | `Literal` |
|
| `PRONOTE_AGENDA_SOURCE` | Source pour l'agenda (`auto`, `ical`, `pronotepy`). | `auto` | `Literal` |
|
||||||
| `PRONOTE_HOMEWORK_SOURCE` | Source pour les devoirs (`auto`, `ical`, `pronotepy`). | `auto` | `Literal` |
|
| `PRONOTE_HOMEWORK_SOURCE` | Source pour les devoirs (`auto`, `ical`, `pronotepy`). | `auto` | `Literal` |
|
||||||
| `PRONOTE_MESSAGES_SOURCE` | Source pour les messages (`pronotepy` uniquement). | `pronotepy` | `Literal` |
|
| `PRONOTE_MESSAGES_SOURCE` | Source pour les messages (`pronotepy` uniquement). | `pronotepy` | `Literal` |
|
||||||
|
| `PRONOTE_URL` | URL de l'instance Pronote. | `None` | `str \| None` |
|
||||||
|
| `PRONOTE_ACCOUNT_TYPE` | Type de compte (`student` ou `parent`). | `parent` | `Literal` |
|
||||||
| `SYNC_PAST_DAYS` | Nombre de jours dans le passé pour la sync CalDAV. | `7` | `int` |
|
| `SYNC_PAST_DAYS` | Nombre de jours dans le passé pour la sync CalDAV. | `7` | `int` |
|
||||||
| `SYNC_FUTURE_DAYS` | Nombre de jours dans le futur pour la sync CalDAV. | `30` | `int` |
|
| `SYNC_FUTURE_DAYS` | Nombre de jours dans le futur pour la sync CalDAV. | `30` | `int` |
|
||||||
|
|
||||||
@@ -309,6 +311,8 @@ PRONOTE_ICAL_URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPL
|
|||||||
PRONOTE_USERNAME=parent.dupont
|
PRONOTE_USERNAME=parent.dupont
|
||||||
PRONOTE_PASSWORD=your_secure_password
|
PRONOTE_PASSWORD=your_secure_password
|
||||||
PRONOTE_ENT=monbureaunumerique
|
PRONOTE_ENT=monbureaunumerique
|
||||||
|
PRONOTE_URL=
|
||||||
|
PRONOTE_ACCOUNT_TYPE=parent
|
||||||
|
|
||||||
# Sources (auto = essayer iCal d'abord, puis pronotepy)
|
# Sources (auto = essayer iCal d'abord, puis pronotepy)
|
||||||
PRONOTE_AGENDA_SOURCE=auto
|
PRONOTE_AGENDA_SOURCE=auto
|
||||||
@@ -369,6 +373,8 @@ class PronoteSettings(BaseSettings):
|
|||||||
username: Optional[str] = None
|
username: Optional[str] = None
|
||||||
password: Optional[SecretStr] = None
|
password: Optional[SecretStr] = None
|
||||||
ent: Optional[str] = None
|
ent: Optional[str] = None
|
||||||
|
pronote_url: Optional[str] = None
|
||||||
|
account_type: Literal["student", "parent"] = "parent"
|
||||||
agenda_source: Literal["auto", "ical", "pronotepy"] = "auto"
|
agenda_source: Literal["auto", "ical", "pronotepy"] = "auto"
|
||||||
homework_source: Literal["auto", "ical", "pronotepy"] = "auto"
|
homework_source: Literal["auto", "ical", "pronotepy"] = "auto"
|
||||||
messages_source: Literal["pronotepy"] = "pronotepy"
|
messages_source: Literal["pronotepy"] = "pronotepy"
|
||||||
@@ -2201,43 +2207,38 @@ class PronoteClient:
|
|||||||
logger.error(f"Échec de la récupération des informations Pronote: {redact_secrets(str(e))}")
|
logger.error(f"Échec de la récupération des informations Pronote: {redact_secrets(str(e))}")
|
||||||
return []
|
return []
|
||||||
|
|
||||||
def get_agenda_fallback(self) -> tuple[List[Lesson], List[HomeworkModel]]:
|
def get_lessons(self, start: date, end: date) -> list[Lesson]:
|
||||||
"""
|
"""
|
||||||
Récupère l'agenda et les devoirs via pronotepy (repli si iCal échoue).
|
Récupère les cours via pronotepy (repli si iCal échoue).
|
||||||
**À utiliser uniquement si PRONOTE_AGENDA_SOURCE=pronotepy ou PRONOTE_HOMEWORK_SOURCE=pronotepy**.
|
Les erreurs sont propagées pour déclencher le repli.
|
||||||
"""
|
"""
|
||||||
try:
|
|
||||||
client = self._get_client()
|
client = self._get_client()
|
||||||
|
return [Lesson(
|
||||||
lessons = []
|
|
||||||
for lesson in client.get_lessons():
|
|
||||||
lessons.append(Lesson(
|
|
||||||
id=str(lesson.id),
|
id=str(lesson.id),
|
||||||
start=lesson.start,
|
start=lesson.start,
|
||||||
end=lesson.end,
|
end=lesson.end,
|
||||||
subject=lesson.subject,
|
subject=lesson.subject,
|
||||||
teachers=[t.name for t in lesson.teachers],
|
teachers=[t.name for t in lesson.teachers],
|
||||||
rooms=[r.name for r in lesson.rooms],
|
rooms=[r.name for r in lesson.rooms],
|
||||||
status=LessonStatus.NORMAL, # À adapter selon les données
|
status=LessonStatus.NORMAL,
|
||||||
content=lesson.content,
|
content=lesson.content,
|
||||||
))
|
) for lesson in client.lessons(start, end)]
|
||||||
|
|
||||||
homeworks = []
|
def get_homeworks(self, start: date, end: date) -> list[HomeworkModel]:
|
||||||
for hw in client.get_homework():
|
"""
|
||||||
homeworks.append(HomeworkModel(
|
Récupère les devoirs via pronotepy.
|
||||||
|
Les erreurs sont propagées pour déclencher le repli.
|
||||||
|
"""
|
||||||
|
client = self._get_client()
|
||||||
|
return [HomeworkModel(
|
||||||
id=str(hw.id),
|
id=str(hw.id),
|
||||||
subject=hw.subject,
|
subject=hw.subject,
|
||||||
teachers=[t.name for t in hw.teachers],
|
teachers=[t.name for t in hw.teachers],
|
||||||
assigned_on=hw.given_date,
|
assigned_on=hw.given_date,
|
||||||
due_on=hw.due_date,
|
due_on=hw.due_date,
|
||||||
text=hw.description,
|
text=hw.description,
|
||||||
html=hw.description, # pronotepy ne fournit pas de HTML
|
html=hw.description,
|
||||||
))
|
) for hw in client.homework(start, end)]
|
||||||
|
|
||||||
return lessons, homeworks
|
|
||||||
except PronoteAPIError as e:
|
|
||||||
logger.error(f"Échec de la récupération de l'agenda via pronotepy: {redact_secrets(str(e))}")
|
|
||||||
return [], []
|
|
||||||
|
|
||||||
def close(self) -> None:
|
def close(self) -> None:
|
||||||
"""Fermeture du client."""
|
"""Fermeture du client."""
|
||||||
@@ -2250,16 +2251,10 @@ class PronoteClient:
|
|||||||
|
|
||||||
```python
|
```python
|
||||||
from typing import Literal, Optional
|
from typing import Literal, Optional
|
||||||
from enum import Enum
|
from enum import StrEnum
|
||||||
from .ical import fetch_ical, parse_ical
|
from .ical import fetch_ical, parse_ical
|
||||||
from .client import PronoteClient
|
from .client import PronoteClient
|
||||||
from ..models.agenda import Lesson, Homework
|
from ..models.agenda import Lesson, SchoolEvent
|
||||||
|
|
||||||
|
|
||||||
class AgendaSource(Enum):
|
|
||||||
AUTO = "auto"
|
|
||||||
ICAL = "ical"
|
|
||||||
PRONOTEPY = "pronotepy"
|
|
||||||
|
|
||||||
|
|
||||||
class PronoteFetcher:
|
class PronoteFetcher:
|
||||||
@@ -2271,6 +2266,8 @@ class PronoteFetcher:
|
|||||||
username: Optional[str] = None,
|
username: Optional[str] = None,
|
||||||
password: Optional[str] = None,
|
password: Optional[str] = None,
|
||||||
ent: Optional[str] = None,
|
ent: Optional[str] = None,
|
||||||
|
pronote_url: Optional[str] = None,
|
||||||
|
account_type: str = "parent",
|
||||||
agenda_source: str = "auto",
|
agenda_source: str = "auto",
|
||||||
homework_source: str = "auto",
|
homework_source: str = "auto",
|
||||||
):
|
):
|
||||||
@@ -2278,6 +2275,8 @@ class PronoteFetcher:
|
|||||||
self.username = username
|
self.username = username
|
||||||
self.password = password
|
self.password = password
|
||||||
self.ent = ent
|
self.ent = ent
|
||||||
|
self.pronote_url = pronote_url
|
||||||
|
self.account_type = account_type
|
||||||
self.agenda_source = AgendaSource(agenda_source)
|
self.agenda_source = AgendaSource(agenda_source)
|
||||||
self.homework_source = AgendaSource(homework_source)
|
self.homework_source = AgendaSource(homework_source)
|
||||||
self._pronote_client: Optional[PronoteClient] = None
|
self._pronote_client: Optional[PronoteClient] = None
|
||||||
@@ -2289,19 +2288,25 @@ class PronoteFetcher:
|
|||||||
password=self.password,
|
password=self.password,
|
||||||
ent=self.ent,
|
ent=self.ent,
|
||||||
ical_url=self.ical_url,
|
ical_url=self.ical_url,
|
||||||
|
pronote_url=self.pronote_url,
|
||||||
|
account_type=self.account_type,
|
||||||
)
|
)
|
||||||
return self._pronote_client
|
return self._pronote_client
|
||||||
|
|
||||||
def fetch_agenda(self) -> tuple[List[Lesson], List[Homework]]:
|
def fetch_agenda(self, start: date, end: date) -> tuple[list[Lesson], list[SchoolEvent]]:
|
||||||
"""Récupère l'agenda selon la source configurée (`agenda_source`)."""
|
"""
|
||||||
|
Récupère l'agenda selon la source configurée (`agenda_source`).
|
||||||
|
En mode AUTO, essaie iCal d'abord, puis pronotepy. Si les deux échouent, lève une erreur explicite.
|
||||||
|
"""
|
||||||
|
from datetime import date
|
||||||
if self.agenda_source == AgendaSource.ICAL:
|
if self.agenda_source == AgendaSource.ICAL:
|
||||||
return self._fetch_agenda_ical()
|
return self._fetch_agenda_ical(start, end)
|
||||||
elif self.agenda_source == AgendaSource.PRONOTEPY:
|
elif self.agenda_source == AgendaSource.PRONOTEPY:
|
||||||
return self._fetch_agenda_pronotepy()
|
return self._fetch_agenda_pronotepy(start, end)
|
||||||
else: # AUTO
|
else: # AUTO
|
||||||
# Essayer iCal d'abord
|
# Essayer iCal d'abord
|
||||||
try:
|
try:
|
||||||
lessons, homeworks = self._fetch_agenda_ical()
|
lessons, homeworks = self._fetch_agenda_ical(start, end)
|
||||||
if lessons or homeworks:
|
if lessons or homeworks:
|
||||||
return lessons, homeworks
|
return lessons, homeworks
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
@@ -2309,30 +2314,33 @@ class PronoteFetcher:
|
|||||||
|
|
||||||
# Repli sur pronotepy
|
# Repli sur pronotepy
|
||||||
logger.info("Repli sur pronotepy pour l'agenda.")
|
logger.info("Repli sur pronotepy pour l'agenda.")
|
||||||
return self._fetch_agenda_pronotepy()
|
return self._fetch_agenda_pronotepy(start, end)
|
||||||
|
|
||||||
def fetch_homework(self) -> List[Homework]:
|
def fetch_homework(self, start: date, end: date) -> list[SchoolEvent]:
|
||||||
"""Récupère les devoirs selon la source configurée (`homework_source`)."""
|
"""
|
||||||
|
Récupère les devoirs selon la source configurée (`homework_source`).
|
||||||
|
En mode AUTO, essaie iCal d'abord, puis pronotepy. Si les deux échouent, lève une erreur explicite.
|
||||||
|
"""
|
||||||
|
from datetime import date
|
||||||
if self.homework_source == AgendaSource.ICAL:
|
if self.homework_source == AgendaSource.ICAL:
|
||||||
# Récupérer uniquement les devoirs depuis iCal
|
# Récupérer uniquement les devoirs depuis iCal
|
||||||
try:
|
try:
|
||||||
_, homeworks = self._fetch_agenda_ical()
|
_, homeworks = self._fetch_agenda_ical(start, end)
|
||||||
return homeworks
|
return homeworks
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning(f"Échec de la récupération iCal pour les devoirs: {redact_secrets(str(e))}")
|
logger.warning(f"Échec de la récupération iCal pour les devoirs: {redact_secrets(str(e))}")
|
||||||
return []
|
raise
|
||||||
elif self.homework_source == AgendaSource.PRONOTEPY:
|
elif self.homework_source == AgendaSource.PRONOTEPY:
|
||||||
# Récupérer uniquement les devoirs depuis pronotepy
|
# Récupérer uniquement les devoirs depuis pronotepy
|
||||||
try:
|
try:
|
||||||
_, homeworks = self._fetch_agenda_pronotepy()
|
return self._fetch_agenda_pronotepy(start, end)[1]
|
||||||
return homeworks
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning(f"Échec de la récupération pronotepy pour les devoirs: {redact_secrets(str(e))}")
|
logger.warning(f"Échec de la récupération pronotepy pour les devoirs: {redact_secrets(str(e))}")
|
||||||
return []
|
raise
|
||||||
else: # AUTO
|
else: # AUTO
|
||||||
# Essayer iCal d'abord
|
# Essayer iCal d'abord
|
||||||
try:
|
try:
|
||||||
_, homeworks = self._fetch_agenda_ical()
|
_, homeworks = self._fetch_agenda_ical(start, end)
|
||||||
if homeworks:
|
if homeworks:
|
||||||
return homeworks
|
return homeworks
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
@@ -2341,33 +2349,32 @@ class PronoteFetcher:
|
|||||||
# Repli sur pronotepy
|
# Repli sur pronotepy
|
||||||
logger.info("Repli sur pronotepy pour les devoirs.")
|
logger.info("Repli sur pronotepy pour les devoirs.")
|
||||||
try:
|
try:
|
||||||
_, homeworks = self._fetch_agenda_pronotepy()
|
return self._fetch_agenda_pronotepy(start, end)[1]
|
||||||
return homeworks
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning(f"Échec de la récupération pronotepy pour les devoirs: {redact_secrets(str(e))}")
|
logger.warning(f"Échec de la récupération pronotepy pour les devoirs: {redact_secrets(str(e))}")
|
||||||
return []
|
raise
|
||||||
|
|
||||||
def _fetch_agenda_ical(self) -> tuple[List[Lesson], List[Homework]]:
|
def _fetch_agenda_ical(self, start: date, end: date) -> tuple[list[Lesson], list[SchoolEvent]]:
|
||||||
"""Récupère l'agenda depuis iCal."""
|
"""Récupère l'agenda depuis iCal."""
|
||||||
if not self.ical_url:
|
if not self.ical_url:
|
||||||
raise ValueError("PRONOTE_ICAL_URL est requis pour la source iCal")
|
raise ValueError("PRONOTE_ICAL_URL est requis pour la source iCal")
|
||||||
|
|
||||||
raw_ical = fetch_ical(self.ical_url)
|
raw_ical = fetch_ical(self.ical_url)
|
||||||
lessons, homeworks, _ = parse_ical(raw_ical)
|
lessons, homeworks, _ = parse_ical(raw_ical, start, end)
|
||||||
return lessons, homeworks
|
return lessons, homeworks
|
||||||
|
|
||||||
def _fetch_agenda_pronotepy(self) -> tuple[List[Lesson], List[Homework]]:
|
def _fetch_agenda_pronotepy(self, start: date, end: date) -> tuple[list[Lesson], list[SchoolEvent]]:
|
||||||
"""Récupère l'agenda depuis pronotepy."""
|
"""Récupère l'agenda depuis pronotepy."""
|
||||||
client = self._get_pronote_client()
|
client = self._get_pronote_client()
|
||||||
lessons, homeworks = client.get_agenda_fallback()
|
lessons = client.get_lessons(start, end)
|
||||||
return lessons, homeworks
|
return lessons, []
|
||||||
|
|
||||||
def fetch_messages(self) -> List[Message]:
|
def fetch_messages(self) -> list[Message]:
|
||||||
"""Récupère les messages (toujours via pronotepy)."""
|
"""Récupère les messages (toujours via pronotepy)."""
|
||||||
client = self._get_pronote_client()
|
client = self._get_pronote_client()
|
||||||
return client.get_messages()
|
return client.get_messages()
|
||||||
|
|
||||||
def fetch_informations(self) -> List[Message]:
|
def fetch_informations(self) -> list[Message]:
|
||||||
"""Récupère les informations (toujours via pronotepy)."""
|
"""Récupère les informations (toujours via pronotepy)."""
|
||||||
client = self._get_pronote_client()
|
client = self._get_pronote_client()
|
||||||
return client.get_informations()
|
return client.get_informations()
|
||||||
@@ -5347,7 +5354,7 @@ def test_parse_ical_homework(parsed_lessons):
|
|||||||
def test_pipeline_full(mock_requests_get, mock_caldav_client, mock_ai_provider, mock_xmpp_channel, sample_settings):
|
def test_pipeline_full(mock_requests_get, mock_caldav_client, mock_ai_provider, mock_xmpp_channel, sample_settings):
|
||||||
"""Test le pipeline complet avec des mocks."""
|
"""Test le pipeline complet avec des mocks."""
|
||||||
from pronote_sync.pipeline.run import PipelineRunner
|
from pronote_sync.pipeline.run import PipelineRunner
|
||||||
from pronote_sync.sources.pronote.fetcher import PronoteFetcher
|
from pronote_sync.sources.pronote.fallback import PronoteFetcher
|
||||||
from pronote_sync.sync.caldav import CalDAVClient
|
from pronote_sync.sync.caldav import CalDAVClient
|
||||||
from pronote_sync.sync.diff import AgendaComparator
|
from pronote_sync.sync.diff import AgendaComparator
|
||||||
from pronote_sync.sources.theoretical.file import CSVTheoreticalAgendaProvider
|
from pronote_sync.sources.theoretical.file import CSVTheoreticalAgendaProvider
|
||||||
|
|||||||
14
TODO.md
14
TODO.md
@@ -73,13 +73,13 @@ Définir tous les modèles de domaine, immuables pour les contrats, mutables pou
|
|||||||
|
|
||||||
Récupérer et normaliser l'agenda, les devoirs et les messages Pronote, avec repli entre iCal et pronotepy.
|
Récupérer et normaliser l'agenda, les devoirs et les messages Pronote, avec repli entre iCal et pronotepy.
|
||||||
|
|
||||||
- [ ] Créer `sources/pronote/ical.py` : `fetch_ical(url)` (HTTP via `requests`, erreurs redactées) et parsing iCal → `Lesson`/`Homework`/`SchoolEvent` (`icalendar`).
|
- [x] Créer `sources/pronote/ical.py` : `fetch_ical(url)` (HTTP via `requests`, erreurs redactées) et parsing iCal → `Lesson`/`Homework`/`SchoolEvent` (`icalendar`).
|
||||||
- [ ] Extraire les blocs de devoirs (`HomeworkBlock`) depuis `DESCRIPTION` et dédupliquer les devoirs (clé normalisée par date).
|
- [x] Extraire les blocs de devoirs (`HomeworkBlock`) depuis `DESCRIPTION` et dédupliquer les devoirs (clé normalisée par date).
|
||||||
- [ ] Détecter les statuts (`CANCELLED`/`MOVED`) via `CATEGORIES` et `STATUS:CANCELLED`.
|
- [x] Détecter les statuts (`CANCELLED`/`MOVED`) via `CATEGORIES` et `STATUS:CANCELLED`.
|
||||||
- [ ] Créer `sources/pronote/client.py` : client `pronotepy` (messages, informations, discussions, sondages, et devoirs en repli) avec masquage des erreurs.
|
- [x] Créer `sources/pronote/client.py` : client `pronotepy` (messages, informations, discussions, sondages, et devoirs en repli) avec masquage des erreurs.
|
||||||
- [ ] Créer `sources/pronote/fallback.py` : sélection de source selon `PRONOTE_*_SOURCE` (auto/ical/pronotepy) et `PronoteFetcher` unifiant `fetch_agenda`/`fetch_homework`/`fetch_messages`.
|
- [x] Créer `sources/pronote/fallback.py` : sélection de source selon `PRONOTE_*_SOURCE` (auto/ical/pronotepy) et `PronoteFetcher` unifiant `fetch_agenda`/`fetch_homework`/`fetch_messages`.
|
||||||
- [ ] Implémenter le repli : iCal échoue → pronotepy ; pronotepy échoue → iCal ; les deux échouent → `PipelineCriticalError`.
|
- [x] Implémenter le repli unique : source primaire + un seul repli (l'autre source si configurée) ; les deux échouent → `PipelineCriticalError`.
|
||||||
- [ ] Normaliser les UID via `utils/uid.normalize_uid` pour la stabilité des événements.
|
- [x] Normaliser les UID via `utils/uid.normalize_pronote_uid` pour la stabilité des événements.
|
||||||
|
|
||||||
### Critères d'acceptation
|
### Critères d'acceptation
|
||||||
- `fetch_ical` parse `tests/fixtures/pronote-4e.ics` en leçons/devoirs/événements corrects (cours annulé détecté).
|
- `fetch_ical` parse `tests/fixtures/pronote-4e.ics` en leçons/devoirs/événements corrects (cours annulé détecté).
|
||||||
|
|||||||
@@ -27,6 +27,8 @@ class PronoteSettings(BaseSettings):
|
|||||||
username: str | None = None
|
username: str | None = None
|
||||||
password: SecretStr | None = None
|
password: SecretStr | None = None
|
||||||
ent: str | None = None
|
ent: str | None = None
|
||||||
|
pronote_url: str | None = None
|
||||||
|
account_type: Literal["student", "parent"] = "parent"
|
||||||
agenda_source: Literal["auto", "ical", "pronotepy"] = "auto"
|
agenda_source: Literal["auto", "ical", "pronotepy"] = "auto"
|
||||||
homework_source: Literal["auto", "ical", "pronotepy"] = "auto"
|
homework_source: Literal["auto", "ical", "pronotepy"] = "auto"
|
||||||
messages_source: Literal["pronotepy"] = "pronotepy"
|
messages_source: Literal["pronotepy"] = "pronotepy"
|
||||||
|
|||||||
33
pronote_sync/errors.py
Normal file
33
pronote_sync/errors.py
Normal file
@@ -0,0 +1,33 @@
|
|||||||
|
"""Hiérarchie d'erreurs transversale du projet pronote-sync."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
|
||||||
|
class PronoteSyncError(Exception):
|
||||||
|
"""Erreur de base pour toutes les exceptions du projet pronote-sync.
|
||||||
|
|
||||||
|
Toutes les exceptions métier héritent de cette classe pour permettre
|
||||||
|
une interception unifiée des erreurs applicatives.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, message: str) -> None:
|
||||||
|
"""Initialise l'erreur avec un message descriptif.
|
||||||
|
|
||||||
|
:param message: Message décrivant la cause de l'erreur.
|
||||||
|
"""
|
||||||
|
super().__init__(message)
|
||||||
|
|
||||||
|
|
||||||
|
class PipelineCriticalError(PronoteSyncError):
|
||||||
|
"""Erreur critique du pipeline, levée quand aucune récupération n'est possible.
|
||||||
|
|
||||||
|
Par exemple : échec simultané des sources iCal et pronotepy,
|
||||||
|
rendant impossible toute synchronisation.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, message: str) -> None:
|
||||||
|
"""Initialise l'erreur critique avec un message descriptif.
|
||||||
|
|
||||||
|
:param message: Message décrivant la cause de l'erreur critique.
|
||||||
|
"""
|
||||||
|
super().__init__(message)
|
||||||
339
pronote_sync/sources/pronote/client.py
Normal file
339
pronote_sync/sources/pronote/client.py
Normal file
@@ -0,0 +1,339 @@
|
|||||||
|
"""Client d'accès à Pronote via ``pronotepy``.
|
||||||
|
|
||||||
|
Ce module fournit l'encapsulation du client ``pronotepy`` pour la source
|
||||||
|
Pronote : récupération des messages des professeurs, des informations et
|
||||||
|
sondages, ainsi que des cours et devoirs en mode repli lorsque le flux
|
||||||
|
iCal échoue. Les erreurs des méthodes dégradées (messages, informations)
|
||||||
|
sont journalisées avec des secrets masqués ; les erreurs de récupération
|
||||||
|
des cours et des devoirs se propagent pour déclencher le repli iCal.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from datetime import date
|
||||||
|
from typing import Any, Protocol
|
||||||
|
|
||||||
|
import pronotepy
|
||||||
|
import pronotepy.ent as pronotepy_ent
|
||||||
|
import requests
|
||||||
|
|
||||||
|
from pronote_sync.config.settings import PronoteSettings
|
||||||
|
from pronote_sync.models.agenda import Lesson, LessonStatus
|
||||||
|
from pronote_sync.models.homework import Homework
|
||||||
|
from pronote_sync.models.message import Message, MessageType
|
||||||
|
from pronote_sync.utils.redaction import redact_exception
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def _get_ent_callable(name: str) -> Any:
|
||||||
|
"""Retourne le callable ``pronotepy`` associé à un nom d'ENT.
|
||||||
|
|
||||||
|
L'accès par :func:`getattr` évite les erreurs ``attr-defined`` de mypy
|
||||||
|
sur les attributs non exportés explicitement par ``pronotepy.ent``.
|
||||||
|
|
||||||
|
:param name: Nom de l'attribut dans ``pronotepy.ent``.
|
||||||
|
:return: Callable ``pronotepy`` associé.
|
||||||
|
:rtype: Any
|
||||||
|
"""
|
||||||
|
return getattr(pronotepy_ent, name)
|
||||||
|
|
||||||
|
|
||||||
|
_ENT_NAMES: list[str] = [
|
||||||
|
"monbureaunumerique",
|
||||||
|
"ent_elyco",
|
||||||
|
"bordeaux",
|
||||||
|
"ent_creuse",
|
||||||
|
"occitanie_montpellier",
|
||||||
|
"paris_classe_numerique",
|
||||||
|
"ile_de_france",
|
||||||
|
"ent_hdf",
|
||||||
|
"ac_orleans_tours",
|
||||||
|
"ac_poitiers",
|
||||||
|
"ac_rennes",
|
||||||
|
"laclasse_educonnect",
|
||||||
|
"ent77",
|
||||||
|
"ent_ecollege78",
|
||||||
|
"ent_essonne",
|
||||||
|
"val_doise",
|
||||||
|
"val_de_marne",
|
||||||
|
"ent_var",
|
||||||
|
"atrium_sud",
|
||||||
|
"laclasse_lyon",
|
||||||
|
"eclat_bfc",
|
||||||
|
"cas_arsene76",
|
||||||
|
"cas_ent27",
|
||||||
|
"cas_kosmos",
|
||||||
|
"ent_creuse_educonnect",
|
||||||
|
"ent_mayotte",
|
||||||
|
"ent_somme",
|
||||||
|
"ent_94",
|
||||||
|
"extranet_colleges_somme",
|
||||||
|
"ac_reunion",
|
||||||
|
]
|
||||||
|
|
||||||
|
_ENT_RESOLVERS: dict[str, Any] = {name: _get_ent_callable(name) for name in _ENT_NAMES}
|
||||||
|
|
||||||
|
|
||||||
|
def _resolve_ent(ent_name: str) -> Any:
|
||||||
|
"""Résout un nom d'ENT en callable ``pronotepy``.
|
||||||
|
|
||||||
|
:param ent_name: Nom de l'ENT tel que configuré (ex. ``"bordeaux"``).
|
||||||
|
:return: Callable ``pronotepy`` associé à l'ENT.
|
||||||
|
:raises ValueError: Si le nom d'ENT n'est pas reconnu.
|
||||||
|
"""
|
||||||
|
resolver = _ENT_RESOLVERS.get(ent_name)
|
||||||
|
if resolver is None:
|
||||||
|
supported = ", ".join(sorted(_ENT_RESOLVERS.keys()))
|
||||||
|
raise ValueError(f"ENT inconnu : {ent_name!r}. ENT supportés : {supported}")
|
||||||
|
return resolver
|
||||||
|
|
||||||
|
|
||||||
|
class PronoteClientProtocol(Protocol):
|
||||||
|
"""Interface du client Pronote consommée par la logique de repli."""
|
||||||
|
|
||||||
|
def get_messages(self) -> list[Message]:
|
||||||
|
"""Récupère les messages des discussions Pronote.
|
||||||
|
|
||||||
|
:return: Liste des messages des professeurs.
|
||||||
|
:rtype: list[Message]
|
||||||
|
"""
|
||||||
|
...
|
||||||
|
|
||||||
|
def get_informations(self) -> list[Message]:
|
||||||
|
"""Récupère les informations et sondages Pronote.
|
||||||
|
|
||||||
|
:return: Liste des informations et sondages.
|
||||||
|
:rtype: list[Message]
|
||||||
|
"""
|
||||||
|
...
|
||||||
|
|
||||||
|
def get_lessons(self, start: date, end: date) -> list[Lesson]:
|
||||||
|
"""Récupère les cours via ``pronotepy`` (repli iCal).
|
||||||
|
|
||||||
|
:param start: Date de début de la fenêtre (incluse).
|
||||||
|
:param end: Date de fin de la fenêtre (incluse).
|
||||||
|
:return: Liste des cours.
|
||||||
|
:rtype: list[Lesson]
|
||||||
|
"""
|
||||||
|
...
|
||||||
|
|
||||||
|
def get_homeworks(self, start: date, end: date) -> list[Homework]:
|
||||||
|
"""Récupère les devoirs via ``pronotepy``.
|
||||||
|
|
||||||
|
:param start: Date de début de la fenêtre (incluse).
|
||||||
|
:param end: Date de fin de la fenêtre (incluse).
|
||||||
|
:return: Liste des devoirs.
|
||||||
|
:rtype: list[Homework]
|
||||||
|
"""
|
||||||
|
...
|
||||||
|
|
||||||
|
|
||||||
|
class PronoteClient:
|
||||||
|
"""Client d'accès à Pronote via ``pronotepy``.
|
||||||
|
|
||||||
|
Encapsule ``pronotepy.Client`` ou ``pronotepy.ParentClient`` selon le
|
||||||
|
type de compte, avec une connexion paresseuse : la connexion n'est
|
||||||
|
établie qu'à la première méthode de récupération appelée. Les erreurs
|
||||||
|
des méthodes dégradées (``get_messages()``, ``get_informations()``)
|
||||||
|
sont journalisées avec des secrets masqués et retournent une valeur
|
||||||
|
vide ; ``get_lessons()`` et ``get_homeworks()`` laissent les
|
||||||
|
exceptions se propager pour déclencher le repli iCal.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, settings: PronoteSettings) -> None:
|
||||||
|
"""Initialise le client Pronote sans se connecter.
|
||||||
|
|
||||||
|
:param settings: Paramètres d'accès à Pronote (username, password, ent).
|
||||||
|
"""
|
||||||
|
self._settings: PronoteSettings = settings
|
||||||
|
self._client: pronotepy.Client | None = None
|
||||||
|
|
||||||
|
def _connect(self) -> pronotepy.Client:
|
||||||
|
"""Crée et connecte le client ``pronotepy`` (connexion paresseuse).
|
||||||
|
|
||||||
|
Le client est créé une seule fois puis réutilisé pour les appels
|
||||||
|
suivants. Le nom d'ENT est résolu via :func:`_resolve_ent` et le
|
||||||
|
type de compte (``student`` ou ``parent``) détermine la classe de
|
||||||
|
client utilisée. L'erreur de connexion est relancée sans
|
||||||
|
journalisation, la méthode publique appelante étant responsable
|
||||||
|
de la journaliser.
|
||||||
|
|
||||||
|
:return: Le client ``pronotepy`` connecté.
|
||||||
|
:rtype: pronotepy.Client
|
||||||
|
:raises ValueError: Si ``pronote_url``, ``username``, ``password``
|
||||||
|
ou ``ent`` est manquant, ou si l'ENT est inconnu.
|
||||||
|
:raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue.
|
||||||
|
"""
|
||||||
|
if self._client is None:
|
||||||
|
pronote_url = self._settings.pronote_url
|
||||||
|
username = self._settings.username
|
||||||
|
password = self._settings.password
|
||||||
|
ent = self._settings.ent
|
||||||
|
if pronote_url is None or username is None or password is None or ent is None:
|
||||||
|
raise ValueError(
|
||||||
|
"pronote_url, username, password et ent sont requis pour pronotepy"
|
||||||
|
)
|
||||||
|
resolver = _resolve_ent(ent)
|
||||||
|
client_class: type[pronotepy.Client] = (
|
||||||
|
pronotepy.ParentClient
|
||||||
|
if self._settings.account_type == "parent"
|
||||||
|
else pronotepy.Client
|
||||||
|
)
|
||||||
|
self._client = client_class(
|
||||||
|
pronote_url=pronote_url,
|
||||||
|
username=username,
|
||||||
|
password=password.get_secret_value(),
|
||||||
|
ent=resolver,
|
||||||
|
)
|
||||||
|
return self._client
|
||||||
|
|
||||||
|
def get_messages(self) -> list[Message]:
|
||||||
|
"""Récupère les messages des discussions Pronote.
|
||||||
|
|
||||||
|
Chaque message d'une discussion est mappé sur un modèle
|
||||||
|
:class:`Message` de type ``DISCUSSION``, le sujet de la discussion
|
||||||
|
servant de titre.
|
||||||
|
|
||||||
|
:return: Liste des messages des professeurs ; vide en cas d'erreur.
|
||||||
|
:rtype: list[Message]
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
client = self._connect()
|
||||||
|
messages: list[Message] = []
|
||||||
|
for discussion in client.discussions():
|
||||||
|
for message in discussion.messages:
|
||||||
|
messages.append(
|
||||||
|
Message(
|
||||||
|
id=message.id,
|
||||||
|
type=MessageType.DISCUSSION,
|
||||||
|
title=discussion.subject,
|
||||||
|
content=message.content,
|
||||||
|
author=message.author or "",
|
||||||
|
date=message.created,
|
||||||
|
read=message.seen,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
return messages
|
||||||
|
except (
|
||||||
|
pronotepy.PronoteAPIError,
|
||||||
|
ValueError,
|
||||||
|
requests.RequestException,
|
||||||
|
ConnectionError,
|
||||||
|
TimeoutError,
|
||||||
|
) as exc:
|
||||||
|
logger.error(
|
||||||
|
"Échec de la récupération des messages Pronote : %s",
|
||||||
|
redact_exception(exc),
|
||||||
|
)
|
||||||
|
return []
|
||||||
|
|
||||||
|
def get_informations(self) -> list[Message]:
|
||||||
|
"""Récupère les informations et sondages Pronote.
|
||||||
|
|
||||||
|
Chaque entrée est mappée sur un modèle :class:`Message` de type
|
||||||
|
``SURVEY`` si c'est un sondage, ``INFORMATION`` sinon.
|
||||||
|
|
||||||
|
:return: Liste des informations et sondages ; vide en cas d'erreur.
|
||||||
|
:rtype: list[Message]
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
client = self._connect()
|
||||||
|
messages: list[Message] = []
|
||||||
|
for info in client.information_and_surveys():
|
||||||
|
messages.append(
|
||||||
|
Message(
|
||||||
|
id=info.id,
|
||||||
|
type=MessageType.SURVEY if info.survey else MessageType.INFORMATION,
|
||||||
|
title=info.title or "",
|
||||||
|
content=info.content(),
|
||||||
|
author=info.author,
|
||||||
|
date=info.creation_date,
|
||||||
|
read=info.read,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
return messages
|
||||||
|
except (
|
||||||
|
pronotepy.PronoteAPIError,
|
||||||
|
ValueError,
|
||||||
|
requests.RequestException,
|
||||||
|
ConnectionError,
|
||||||
|
TimeoutError,
|
||||||
|
) as exc:
|
||||||
|
logger.error(
|
||||||
|
"Échec de la récupération des informations Pronote : %s",
|
||||||
|
redact_exception(exc),
|
||||||
|
)
|
||||||
|
return []
|
||||||
|
|
||||||
|
def get_lessons(self, start: date, end: date) -> list[Lesson]:
|
||||||
|
"""Récupère les cours via ``pronotepy`` (repli iCal).
|
||||||
|
|
||||||
|
Les exceptions ne sont pas attrapées : elles se propagent afin que
|
||||||
|
l'appelant puisse détecter l'échec et déclencher le repli (ou une
|
||||||
|
erreur explicite).
|
||||||
|
|
||||||
|
:param start: Date de début de la fenêtre (incluse).
|
||||||
|
:param end: Date de fin de la fenêtre (incluse).
|
||||||
|
:return: Liste des cours.
|
||||||
|
:rtype: list[Lesson]
|
||||||
|
:raises pronotepy.PronoteAPIError: Si l'API Pronote échoue.
|
||||||
|
:raises ValueError: Si la configuration ou l'ENT est invalide.
|
||||||
|
:raises requests.RequestException: Si une requête réseau échoue.
|
||||||
|
:raises ConnectionError: Si la connexion réseau échoue.
|
||||||
|
:raises TimeoutError: Si la requête réseau expire.
|
||||||
|
"""
|
||||||
|
client = self._connect()
|
||||||
|
lessons: list[Lesson] = []
|
||||||
|
for lesson in client.lessons(start, end):
|
||||||
|
content = lesson.content
|
||||||
|
lessons.append(
|
||||||
|
Lesson(
|
||||||
|
id=lesson.id,
|
||||||
|
start=lesson.start,
|
||||||
|
end=lesson.end,
|
||||||
|
subject=lesson.subject.name if lesson.subject is not None else "",
|
||||||
|
teachers=tuple(lesson.teacher_names or ()),
|
||||||
|
rooms=tuple(lesson.classrooms or ()),
|
||||||
|
group=lesson.group_name,
|
||||||
|
status=(LessonStatus.CANCELLED if lesson.canceled else LessonStatus.NORMAL),
|
||||||
|
content=content.description if content is not None else None,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
return lessons
|
||||||
|
|
||||||
|
def get_homeworks(self, start: date, end: date) -> list[Homework]:
|
||||||
|
"""Récupère les devoirs via ``pronotepy``.
|
||||||
|
|
||||||
|
Les exceptions ne sont pas attrapées : elles se propagent afin que
|
||||||
|
l'appelant puisse détecter l'échec et déclencher le repli (ou une
|
||||||
|
erreur explicite). **pronotepy** ne fournissant ni la date de
|
||||||
|
distribution ni les professeurs des devoirs, ces champs restent
|
||||||
|
vides.
|
||||||
|
|
||||||
|
:param start: Date de début de la fenêtre (incluse).
|
||||||
|
:param end: Date de fin de la fenêtre (incluse).
|
||||||
|
:return: Liste des devoirs.
|
||||||
|
:rtype: list[Homework]
|
||||||
|
:raises pronotepy.PronoteAPIError: Si l'API Pronote échoue.
|
||||||
|
:raises ValueError: Si la configuration ou l'ENT est invalide.
|
||||||
|
:raises requests.RequestException: Si une requête réseau échoue.
|
||||||
|
:raises ConnectionError: Si la connexion réseau échoue.
|
||||||
|
:raises TimeoutError: Si la requête réseau expire.
|
||||||
|
"""
|
||||||
|
client = self._connect()
|
||||||
|
homeworks: list[Homework] = []
|
||||||
|
for hw in client.homework(start, end):
|
||||||
|
homeworks.append(
|
||||||
|
Homework(
|
||||||
|
id=hw.id,
|
||||||
|
subject=hw.subject.name,
|
||||||
|
teachers=(),
|
||||||
|
assigned_on=None,
|
||||||
|
due_on=hw.date,
|
||||||
|
text=hw.description,
|
||||||
|
html=hw.description,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
return homeworks
|
||||||
405
pronote_sync/sources/pronote/fallback.py
Normal file
405
pronote_sync/sources/pronote/fallback.py
Normal file
@@ -0,0 +1,405 @@
|
|||||||
|
"""Logique de repli unifiant les sources iCal et pronotepy.
|
||||||
|
|
||||||
|
Ce module fournit l'enum :class:`AgendaSource`, le protocole
|
||||||
|
:class:`PronoteFetcherProtocol` consommé par le pipeline ainsi que la
|
||||||
|
classe :class:`PronoteFetcher` qui sélectionne la source selon la
|
||||||
|
configuration (``PRONOTE_AGENDA_SOURCE`` / ``PRONOTE_HOMEWORK_SOURCE``).
|
||||||
|
La logique applique un repli **unique** : une source primaire est
|
||||||
|
essayée en premier et, en cas d'échec, une seule source de repli (jamais
|
||||||
|
réciproque ni itératif) est essayée si elle est configurée. Les messages
|
||||||
|
et informations proviennent toujours de pronotepy. Toutes les erreurs
|
||||||
|
sont journalisées avec des secrets masqués via
|
||||||
|
:func:`~pronote_sync.utils.redaction.redact_exception` ; les exceptions
|
||||||
|
d'origine ne sont jamais chaînées (``from None``).
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from datetime import date, timedelta
|
||||||
|
from enum import StrEnum
|
||||||
|
from typing import Literal, Protocol
|
||||||
|
|
||||||
|
from pronote_sync.config.settings import Settings
|
||||||
|
from pronote_sync.errors import PipelineCriticalError
|
||||||
|
from pronote_sync.models.agenda import Lesson, SchoolEvent
|
||||||
|
from pronote_sync.models.homework import Homework
|
||||||
|
from pronote_sync.models.message import Message
|
||||||
|
from pronote_sync.sources.pronote.client import PronoteClientProtocol
|
||||||
|
from pronote_sync.sources.pronote.ical import collect_homeworks, fetch_ical, parse_ical
|
||||||
|
from pronote_sync.utils.redaction import redact_exception
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
_SourceName = Literal["ical", "pronotepy"]
|
||||||
|
|
||||||
|
|
||||||
|
class AgendaSource(StrEnum):
|
||||||
|
"""Source configurée pour la récupération de l'agenda et des devoirs."""
|
||||||
|
|
||||||
|
AUTO = "auto"
|
||||||
|
ICAL = "ical"
|
||||||
|
PRONOTEPY = "pronotepy"
|
||||||
|
|
||||||
|
|
||||||
|
class PronoteFetcherProtocol(Protocol):
|
||||||
|
"""Interface de récupération des données Pronote consommée par le pipeline."""
|
||||||
|
|
||||||
|
def fetch_agenda(self) -> tuple[list[Lesson], list[SchoolEvent]]:
|
||||||
|
"""Récupère les cours et les événements scolaires.
|
||||||
|
|
||||||
|
:return: Tuple ``(cours, événements scolaires)``.
|
||||||
|
:rtype: tuple[list[Lesson], list[SchoolEvent]]
|
||||||
|
:raises PipelineCriticalError: Si toutes les sources configurées échouent.
|
||||||
|
"""
|
||||||
|
...
|
||||||
|
|
||||||
|
def fetch_homework(self, target_date: date) -> list[Homework]:
|
||||||
|
"""Récupère les devoirs pour la date cible.
|
||||||
|
|
||||||
|
:param target_date: Date cible pour laquelle collecter les devoirs.
|
||||||
|
:return: Liste des devoirs.
|
||||||
|
:rtype: list[Homework]
|
||||||
|
:raises PipelineCriticalError: Si toutes les sources configurées échouent.
|
||||||
|
"""
|
||||||
|
...
|
||||||
|
|
||||||
|
def fetch_messages(self) -> list[Message]:
|
||||||
|
"""Récupère les messages des discussions Pronote.
|
||||||
|
|
||||||
|
:return: Liste des messages.
|
||||||
|
:rtype: list[Message]
|
||||||
|
"""
|
||||||
|
...
|
||||||
|
|
||||||
|
def fetch_informations(self) -> list[Message]:
|
||||||
|
"""Récupère les informations et sondages Pronote.
|
||||||
|
|
||||||
|
:return: Liste des informations et sondages.
|
||||||
|
:rtype: list[Message]
|
||||||
|
"""
|
||||||
|
...
|
||||||
|
|
||||||
|
|
||||||
|
class PronoteFetcher:
|
||||||
|
"""Récupère les données Pronote via iCal ou pronotepy avec repli unique.
|
||||||
|
|
||||||
|
Unifie les sources iCal et pronotepy selon la source configurée
|
||||||
|
(``agenda_source`` / ``homework_source``) : la source primaire est
|
||||||
|
essayée en premier et, si elle échoue, une seule source de repli est
|
||||||
|
essayée lorsqu'elle est configurée. Les messages et informations
|
||||||
|
proviennent toujours de pronotepy.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, settings: Settings, pronote_client: PronoteClientProtocol) -> None:
|
||||||
|
"""Initialise le fetcher sans récupérer aucune donnée.
|
||||||
|
|
||||||
|
:param settings: Configuration racine du pipeline (Pronote, fenêtre de synchronisation).
|
||||||
|
:param pronote_client: Client pronotepy utilisé pour les sources pronotepy.
|
||||||
|
"""
|
||||||
|
self._settings: Settings = settings
|
||||||
|
self._pronote_client: PronoteClientProtocol = pronote_client
|
||||||
|
|
||||||
|
def _fetch_window(self) -> tuple[date, date]:
|
||||||
|
"""Calcule la fenêtre de synchronisation autour de la date du jour.
|
||||||
|
|
||||||
|
:return: Tuple ``(date de début, date de fin)`` de la fenêtre.
|
||||||
|
:rtype: tuple[date, date]
|
||||||
|
"""
|
||||||
|
today = date.today()
|
||||||
|
start = today - timedelta(days=self._settings.app.sync_past_days)
|
||||||
|
end = today + timedelta(days=self._settings.app.sync_future_days)
|
||||||
|
return start, end
|
||||||
|
|
||||||
|
def _is_ical_configured(self) -> bool:
|
||||||
|
"""Vérifie que la source iCal est configurée.
|
||||||
|
|
||||||
|
:return: ``True`` si ``ical_url`` est défini, ``False`` sinon.
|
||||||
|
:rtype: bool
|
||||||
|
"""
|
||||||
|
return self._settings.pronote.ical_url is not None
|
||||||
|
|
||||||
|
def _is_pronotepy_configured(self) -> bool:
|
||||||
|
"""Vérifie que la source pronotepy est entièrement configurée.
|
||||||
|
|
||||||
|
:return: ``True`` si ``pronote_url``, ``username``, ``password``
|
||||||
|
et ``ent`` sont tous définis, ``False`` sinon.
|
||||||
|
:rtype: bool
|
||||||
|
"""
|
||||||
|
pronote = self._settings.pronote
|
||||||
|
return (
|
||||||
|
pronote.pronote_url is not None
|
||||||
|
and pronote.username is not None
|
||||||
|
and pronote.password is not None
|
||||||
|
and pronote.ent is not None
|
||||||
|
)
|
||||||
|
|
||||||
|
def _fetch_agenda_ical(self) -> tuple[list[Lesson], list[SchoolEvent]]:
|
||||||
|
"""Récupère l'agenda depuis le flux iCal.
|
||||||
|
|
||||||
|
:return: Tuple ``(cours, événements scolaires)``.
|
||||||
|
:rtype: tuple[list[Lesson], list[SchoolEvent]]
|
||||||
|
:raises ValueError: Si ``ical_url`` n'est pas configuré ou si le flux est invalide.
|
||||||
|
:raises OSError: Si le fichier iCal local est illisible.
|
||||||
|
:raises requests.RequestException: Si la récupération HTTP échoue.
|
||||||
|
"""
|
||||||
|
ical_url = self._settings.pronote.ical_url
|
||||||
|
if ical_url is None:
|
||||||
|
raise ValueError("PRONOTE_ICAL_URL est requis pour la source iCal")
|
||||||
|
raw_ical = fetch_ical(ical_url.get_secret_value())
|
||||||
|
lessons, _, school_events = parse_ical(raw_ical)
|
||||||
|
return lessons, school_events
|
||||||
|
|
||||||
|
def _fetch_agenda_pronotepy(self) -> tuple[list[Lesson], list[SchoolEvent]]:
|
||||||
|
"""Récupère l'agenda depuis pronotepy.
|
||||||
|
|
||||||
|
Les événements scolaires (vacances, jours fériés) ne sont pas
|
||||||
|
fournis par pronotepy : la liste retournée est vide.
|
||||||
|
|
||||||
|
:return: Tuple ``(cours, événements scolaires)``.
|
||||||
|
:rtype: tuple[list[Lesson], list[SchoolEvent]]
|
||||||
|
:raises pronotepy.PronoteAPIError: Si l'API Pronote échoue.
|
||||||
|
:raises ValueError: Si la configuration ou l'ENT est invalide.
|
||||||
|
:raises requests.RequestException: Si une requête réseau échoue.
|
||||||
|
:raises ConnectionError: Si la connexion réseau échoue.
|
||||||
|
:raises TimeoutError: Si la requête réseau expire.
|
||||||
|
"""
|
||||||
|
start, end = self._fetch_window()
|
||||||
|
lessons = self._pronote_client.get_lessons(start, end)
|
||||||
|
return lessons, []
|
||||||
|
|
||||||
|
def _agenda_sources(self) -> tuple[_SourceName, _SourceName | None]:
|
||||||
|
"""Sélectionne la source primaire et le repli unique pour l'agenda.
|
||||||
|
|
||||||
|
En mode ``AUTO``, iCal est primaire si ``ical_url`` est configuré
|
||||||
|
(repli pronotepy si la configuration pronotepy est complète),
|
||||||
|
sinon pronotepy sans repli. Les modes explicites ``ICAL`` et
|
||||||
|
``PRONOTEPY`` désignent la source primaire et tentent l'autre
|
||||||
|
source en repli unique si elle est configurée.
|
||||||
|
|
||||||
|
:return: Tuple ``(source primaire, source de repli ou ``None``)``.
|
||||||
|
:rtype: tuple[_SourceName, _SourceName | None]
|
||||||
|
:raises PipelineCriticalError: Si aucune source n'est configurée en mode ``AUTO``.
|
||||||
|
"""
|
||||||
|
source = AgendaSource(self._settings.pronote.agenda_source)
|
||||||
|
if source is AgendaSource.ICAL:
|
||||||
|
return "ical", "pronotepy" if self._is_pronotepy_configured() else None
|
||||||
|
if source is AgendaSource.PRONOTEPY:
|
||||||
|
return "pronotepy", "ical" if self._is_ical_configured() else None
|
||||||
|
if self._is_ical_configured():
|
||||||
|
return "ical", "pronotepy" if self._is_pronotepy_configured() else None
|
||||||
|
if self._is_pronotepy_configured():
|
||||||
|
return "pronotepy", None
|
||||||
|
raise PipelineCriticalError(
|
||||||
|
"Impossible de récupérer l'agenda : ni la source iCal ni pronotepy n'est configurée"
|
||||||
|
) from None
|
||||||
|
|
||||||
|
def _fetch_agenda_source(self, name: _SourceName) -> tuple[list[Lesson], list[SchoolEvent]]:
|
||||||
|
"""Récupère l'agenda depuis la source nommée.
|
||||||
|
|
||||||
|
:param name: Nom de la source (``"ical"`` ou ``"pronotepy"``).
|
||||||
|
:return: Tuple ``(cours, événements scolaires)``.
|
||||||
|
:rtype: tuple[list[Lesson], list[SchoolEvent]]
|
||||||
|
"""
|
||||||
|
if name == "ical":
|
||||||
|
return self._fetch_agenda_ical()
|
||||||
|
return self._fetch_agenda_pronotepy()
|
||||||
|
|
||||||
|
def fetch_agenda(self) -> tuple[list[Lesson], list[SchoolEvent]]:
|
||||||
|
"""Récupère les cours et les événements scolaires selon la source configurée.
|
||||||
|
|
||||||
|
La source primaire est essayée en premier ; si elle échoue, la
|
||||||
|
source de repli unique (l'autre source, si configurée) est
|
||||||
|
essayée. Si la source primaire et le repli échouent — ou si
|
||||||
|
aucune source n'est configurée en mode ``AUTO`` — une erreur
|
||||||
|
critique est levée.
|
||||||
|
|
||||||
|
:return: Tuple ``(cours, événements scolaires)``.
|
||||||
|
:rtype: tuple[list[Lesson], list[SchoolEvent]]
|
||||||
|
:raises PipelineCriticalError: Si toutes les sources tentées échouent.
|
||||||
|
"""
|
||||||
|
primary, fallback = self._agenda_sources()
|
||||||
|
try:
|
||||||
|
return self._fetch_agenda_source(primary)
|
||||||
|
except Exception as exc:
|
||||||
|
logger.error(
|
||||||
|
"Échec de la récupération %s pour l'agenda : %s",
|
||||||
|
primary,
|
||||||
|
redact_exception(exc),
|
||||||
|
)
|
||||||
|
if fallback is None:
|
||||||
|
raise PipelineCriticalError(
|
||||||
|
f"Impossible de récupérer l'agenda : la source {primary} a échoué"
|
||||||
|
) from None
|
||||||
|
logger.info("Repli sur %s pour l'agenda.", fallback)
|
||||||
|
try:
|
||||||
|
lessons, school_events = self._fetch_agenda_source(fallback)
|
||||||
|
except Exception as exc:
|
||||||
|
logger.error(
|
||||||
|
"Échec de la récupération %s pour l'agenda : %s",
|
||||||
|
fallback,
|
||||||
|
redact_exception(exc),
|
||||||
|
)
|
||||||
|
raise PipelineCriticalError(
|
||||||
|
f"Impossible de récupérer l'agenda : les sources {primary}"
|
||||||
|
f" et {fallback} ont échoué"
|
||||||
|
) from None
|
||||||
|
if not lessons:
|
||||||
|
logger.warning(
|
||||||
|
"Le repli %s pour l'agenda a retourné un résultat vide après l'échec "
|
||||||
|
"de %s : impossible de distinguer une absence de cours d'un échec "
|
||||||
|
"silencieux.",
|
||||||
|
fallback,
|
||||||
|
primary,
|
||||||
|
)
|
||||||
|
return lessons, school_events
|
||||||
|
|
||||||
|
def _fetch_homework_ical(self, target_date: date) -> list[Homework]:
|
||||||
|
"""Récupère les devoirs depuis le flux iCal pour la date cible.
|
||||||
|
|
||||||
|
:param target_date: Date cible pour laquelle collecter les devoirs.
|
||||||
|
:return: Liste des devoirs.
|
||||||
|
:rtype: list[Homework]
|
||||||
|
:raises ValueError: Si ``ical_url`` n'est pas configuré ou si le flux est invalide.
|
||||||
|
:raises OSError: Si le fichier iCal local est illisible.
|
||||||
|
:raises requests.RequestException: Si la récupération HTTP échoue.
|
||||||
|
"""
|
||||||
|
lessons, _ = self._fetch_agenda_ical()
|
||||||
|
return collect_homeworks(lessons, target_date)
|
||||||
|
|
||||||
|
def _fetch_homework_pronotepy(self, target_date: date) -> list[Homework]:
|
||||||
|
"""Récupère les devoirs depuis pronotepy pour la date cible.
|
||||||
|
|
||||||
|
Les devoirs sont filtrés sur la date d'échéance : seuls ceux dont
|
||||||
|
``due_on`` correspond à ``target_date`` sont conservés.
|
||||||
|
|
||||||
|
:param target_date: Date cible pour laquelle collecter les devoirs.
|
||||||
|
:return: Liste des devoirs.
|
||||||
|
:rtype: list[Homework]
|
||||||
|
:raises pronotepy.PronoteAPIError: Si l'API Pronote échoue.
|
||||||
|
:raises ValueError: Si la configuration ou l'ENT est invalide.
|
||||||
|
:raises requests.RequestException: Si une requête réseau échoue.
|
||||||
|
:raises ConnectionError: Si la connexion réseau échoue.
|
||||||
|
:raises TimeoutError: Si la requête réseau expire.
|
||||||
|
"""
|
||||||
|
start, end = self._fetch_window()
|
||||||
|
homeworks = self._pronote_client.get_homeworks(start, end)
|
||||||
|
return [hw for hw in homeworks if hw.due_on == target_date]
|
||||||
|
|
||||||
|
def _homework_sources(self) -> tuple[_SourceName, _SourceName | None]:
|
||||||
|
"""Sélectionne la source primaire et le repli unique pour les devoirs.
|
||||||
|
|
||||||
|
En mode ``AUTO``, iCal est primaire si ``ical_url`` est configuré
|
||||||
|
(repli pronotepy si la configuration pronotepy est complète),
|
||||||
|
sinon pronotepy sans repli. Les modes explicites ``ICAL`` et
|
||||||
|
``PRONOTEPY`` désignent la source primaire et tentent l'autre
|
||||||
|
source en repli unique si elle est configurée.
|
||||||
|
|
||||||
|
:return: Tuple ``(source primaire, source de repli ou ``None``)``.
|
||||||
|
:rtype: tuple[_SourceName, _SourceName | None]
|
||||||
|
:raises PipelineCriticalError: Si aucune source n'est configurée en mode ``AUTO``.
|
||||||
|
"""
|
||||||
|
source = AgendaSource(self._settings.pronote.homework_source)
|
||||||
|
if source is AgendaSource.ICAL:
|
||||||
|
return "ical", "pronotepy" if self._is_pronotepy_configured() else None
|
||||||
|
if source is AgendaSource.PRONOTEPY:
|
||||||
|
return "pronotepy", "ical" if self._is_ical_configured() else None
|
||||||
|
if self._is_ical_configured():
|
||||||
|
return "ical", "pronotepy" if self._is_pronotepy_configured() else None
|
||||||
|
if self._is_pronotepy_configured():
|
||||||
|
return "pronotepy", None
|
||||||
|
raise PipelineCriticalError(
|
||||||
|
"Impossible de récupérer les devoirs : ni la source iCal ni pronotepy n'est configurée"
|
||||||
|
) from None
|
||||||
|
|
||||||
|
def _fetch_homework_source(self, name: _SourceName, target_date: date) -> list[Homework]:
|
||||||
|
"""Récupère les devoirs depuis la source nommée.
|
||||||
|
|
||||||
|
:param name: Nom de la source (``"ical"`` ou ``"pronotepy"``).
|
||||||
|
:param target_date: Date cible pour laquelle collecter les devoirs.
|
||||||
|
:return: Liste des devoirs.
|
||||||
|
:rtype: list[Homework]
|
||||||
|
"""
|
||||||
|
if name == "ical":
|
||||||
|
return self._fetch_homework_ical(target_date)
|
||||||
|
return self._fetch_homework_pronotepy(target_date)
|
||||||
|
|
||||||
|
def fetch_homework(self, target_date: date) -> list[Homework]:
|
||||||
|
"""Récupère les devoirs selon la source configurée.
|
||||||
|
|
||||||
|
La source primaire est essayée en premier ; si elle échoue, la
|
||||||
|
source de repli unique (l'autre source, si configurée) est
|
||||||
|
essayée. Si la source primaire et le repli échouent — ou si
|
||||||
|
aucune source n'est configurée en mode ``AUTO`` — une erreur
|
||||||
|
critique est levée.
|
||||||
|
|
||||||
|
:param target_date: Date cible pour laquelle collecter les devoirs.
|
||||||
|
:return: Liste des devoirs.
|
||||||
|
:rtype: list[Homework]
|
||||||
|
:raises PipelineCriticalError: Si toutes les sources tentées échouent.
|
||||||
|
"""
|
||||||
|
primary, fallback = self._homework_sources()
|
||||||
|
try:
|
||||||
|
return self._fetch_homework_source(primary, target_date)
|
||||||
|
except Exception as exc:
|
||||||
|
logger.error(
|
||||||
|
"Échec de la récupération %s pour les devoirs : %s",
|
||||||
|
primary,
|
||||||
|
redact_exception(exc),
|
||||||
|
)
|
||||||
|
if fallback is None:
|
||||||
|
raise PipelineCriticalError(
|
||||||
|
f"Impossible de récupérer les devoirs : la source {primary} a échoué"
|
||||||
|
) from None
|
||||||
|
logger.info("Repli sur %s pour les devoirs.", fallback)
|
||||||
|
try:
|
||||||
|
homeworks = self._fetch_homework_source(fallback, target_date)
|
||||||
|
except Exception as exc:
|
||||||
|
logger.error(
|
||||||
|
"Échec de la récupération %s pour les devoirs : %s",
|
||||||
|
fallback,
|
||||||
|
redact_exception(exc),
|
||||||
|
)
|
||||||
|
raise PipelineCriticalError(
|
||||||
|
f"Impossible de récupérer les devoirs : les sources {primary}"
|
||||||
|
f" et {fallback} ont échoué"
|
||||||
|
) from None
|
||||||
|
if not homeworks:
|
||||||
|
logger.warning(
|
||||||
|
"Le repli %s pour les devoirs a retourné un résultat vide après "
|
||||||
|
"l'échec de %s : impossible de distinguer une absence de devoirs "
|
||||||
|
"d'un échec silencieux.",
|
||||||
|
fallback,
|
||||||
|
primary,
|
||||||
|
)
|
||||||
|
return homeworks
|
||||||
|
|
||||||
|
def fetch_messages(self) -> list[Message]:
|
||||||
|
"""Récupère les messages des discussions Pronote (toujours via pronotepy).
|
||||||
|
|
||||||
|
:return: Liste des messages.
|
||||||
|
:rtype: list[Message]
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
return self._pronote_client.get_messages()
|
||||||
|
except Exception as exc:
|
||||||
|
logger.error(
|
||||||
|
"Échec de la récupération des messages : %s",
|
||||||
|
redact_exception(exc),
|
||||||
|
)
|
||||||
|
raise
|
||||||
|
|
||||||
|
def fetch_informations(self) -> list[Message]:
|
||||||
|
"""Récupère les informations et sondages Pronote (toujours via pronotepy).
|
||||||
|
|
||||||
|
:return: Liste des informations et sondages.
|
||||||
|
:rtype: list[Message]
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
return self._pronote_client.get_informations()
|
||||||
|
except Exception as exc:
|
||||||
|
logger.error(
|
||||||
|
"Échec de la récupération des informations : %s",
|
||||||
|
redact_exception(exc),
|
||||||
|
)
|
||||||
|
raise
|
||||||
467
pronote_sync/sources/pronote/ical.py
Normal file
467
pronote_sync/sources/pronote/ical.py
Normal file
@@ -0,0 +1,467 @@
|
|||||||
|
"""Récupération et parsing du flux iCal Pronote.
|
||||||
|
|
||||||
|
Ce module fournit le téléchargement du flux iCal Pronote (via HTTP ou
|
||||||
|
``file://`` pour les tests) ainsi que son parsing en modèles : cours
|
||||||
|
(:class:`~pronote_sync.models.agenda.Lesson`), événements scolaires
|
||||||
|
(:class:`~pronote_sync.models.agenda.SchoolEvent`) et devoirs
|
||||||
|
(:class:`~pronote_sync.models.homework.Homework`).
|
||||||
|
|
||||||
|
La collecte finale des devoirs est réalisée à part, une fois la date
|
||||||
|
cible connue, via :func:`collect_homeworks`.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import hashlib
|
||||||
|
import re
|
||||||
|
import urllib.parse
|
||||||
|
from datetime import date, datetime
|
||||||
|
from html import unescape
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import TypedDict
|
||||||
|
|
||||||
|
import requests
|
||||||
|
from icalendar import Calendar
|
||||||
|
|
||||||
|
from ...models.agenda import (
|
||||||
|
HomeworkBlock,
|
||||||
|
Lesson,
|
||||||
|
LessonStatus,
|
||||||
|
SchoolEvent,
|
||||||
|
SchoolEventKind,
|
||||||
|
)
|
||||||
|
from ...models.homework import Homework
|
||||||
|
from ...utils.redaction import redact_exception, redact_url
|
||||||
|
from ...utils.uid import generate_deterministic_uid, normalize_pronote_uid
|
||||||
|
|
||||||
|
_HEADER_LABEL_PATTERN = re.compile(r"\b(Matière|Professeurs?|Salles?|Groupe)\s*:\s*")
|
||||||
|
_CALNAME_PATTERN = re.compile(r"^X-WR-CALNAME(?:;[^:]*)?:([^\r\n]*)", re.MULTILINE)
|
||||||
|
_TAG_PATTERN = re.compile(r"<[^>]+>")
|
||||||
|
_CONTENT_PATTERN = re.compile(
|
||||||
|
r"<strong>Contenu pédagogique\s*:\s*</strong>(.*?)(?=<strong>|</div>\s*$|\Z)",
|
||||||
|
re.DOTALL,
|
||||||
|
)
|
||||||
|
_DUE_PATTERN = re.compile(
|
||||||
|
r"<strong>Pour le (\d{2}/\d{2}/\d{4})\s*:\s*</strong>(.*?)(?=<strong>|</div>\s*$|\Z)",
|
||||||
|
re.DOTALL,
|
||||||
|
)
|
||||||
|
_ASSIGNED_PATTERN = re.compile(
|
||||||
|
r"<strong>Donné le (\d{2}/\d{2}/\d{4})\s*:\s*</strong>(.*?)(?=<strong>|</div>\s*$|\Z)",
|
||||||
|
re.DOTALL,
|
||||||
|
)
|
||||||
|
|
||||||
|
_HEADERS = {
|
||||||
|
"accept": "text/calendar",
|
||||||
|
"user-agent": "pronote-sync",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
class HeaderInfo(TypedDict):
|
||||||
|
"""Métadonnées du cours extraites de l'en-tête de la DESCRIPTION."""
|
||||||
|
|
||||||
|
subject: str
|
||||||
|
teachers: list[str]
|
||||||
|
rooms: list[str]
|
||||||
|
group: str | None
|
||||||
|
|
||||||
|
|
||||||
|
def fetch_ical(url: str, timeout: int = 20) -> str:
|
||||||
|
"""Récupère le contenu brut d'un flux iCal Pronote.
|
||||||
|
|
||||||
|
Gère les URLs ``file://`` pour les tests locaux (le chemin est
|
||||||
|
décodé de l'échappement URI, ex. ``%20`` → espace) et valide que
|
||||||
|
le flux commence bien par ``BEGIN:VCALENDAR``. Toutes les erreurs
|
||||||
|
sont relancées avec un message dont les secrets (token
|
||||||
|
``icalsecurise``) sont masqués.
|
||||||
|
|
||||||
|
:param url: URL du flux iCal (avec token ``icalsecurise``) ou chemin ``file://``.
|
||||||
|
:param timeout: Timeout HTTP en secondes (défaut : 20).
|
||||||
|
:return: Contenu brut du flux iCal.
|
||||||
|
:rtype: str
|
||||||
|
:raises OSError: Si le fichier local ``file://`` est illisible.
|
||||||
|
:raises requests.RequestException: Si la récupération HTTP échoue.
|
||||||
|
:raises ValueError: Si le flux ne commence pas par ``BEGIN:VCALENDAR``.
|
||||||
|
"""
|
||||||
|
if url.startswith("file://"):
|
||||||
|
parsed_url = urllib.parse.urlparse(url)
|
||||||
|
path = Path(urllib.parse.unquote(parsed_url.path))
|
||||||
|
try:
|
||||||
|
content = path.read_text(encoding="utf-8")
|
||||||
|
except OSError as exc:
|
||||||
|
raise OSError(
|
||||||
|
f"Impossible de lire le fichier iCal {redact_url(url)} : {redact_exception(exc)}"
|
||||||
|
) from exc
|
||||||
|
if not content.lstrip().startswith("BEGIN:VCALENDAR"):
|
||||||
|
raise ValueError(f"Fichier iCal invalide (pas de BEGIN:VCALENDAR) : {redact_url(url)}")
|
||||||
|
return content
|
||||||
|
|
||||||
|
try:
|
||||||
|
response = requests.get(url, headers=_HEADERS, timeout=timeout)
|
||||||
|
response.raise_for_status()
|
||||||
|
content = response.text
|
||||||
|
except Exception as exc:
|
||||||
|
raise requests.RequestException(
|
||||||
|
f"Échec de la récupération du flux iCal {redact_url(url)} : {redact_exception(exc)}"
|
||||||
|
) from exc
|
||||||
|
|
||||||
|
if not content.lstrip().startswith("BEGIN:VCALENDAR"):
|
||||||
|
raise ValueError(f"Flux iCal invalide (pas de BEGIN:VCALENDAR) : {redact_url(url)}")
|
||||||
|
return str(content)
|
||||||
|
|
||||||
|
|
||||||
|
def _unfold_ical(raw_ical: str) -> str:
|
||||||
|
"""Déplie les lignes de continuation iCalendar.
|
||||||
|
|
||||||
|
Une ligne commençant par un espace ou une tabulation prolonge la
|
||||||
|
ligne précédente : le caractère d'espacement initial est retiré et
|
||||||
|
la suite est jointe à la ligne précédente.
|
||||||
|
|
||||||
|
:param raw_ical: Contenu brut du flux iCal.
|
||||||
|
:return: Contenu avec les lignes de continuation dépliées.
|
||||||
|
:rtype: str
|
||||||
|
"""
|
||||||
|
unfolded: list[str] = []
|
||||||
|
for line in raw_ical.splitlines():
|
||||||
|
if line.startswith((" ", "\t")) and unfolded:
|
||||||
|
unfolded[-1] += line[1:]
|
||||||
|
else:
|
||||||
|
unfolded.append(line)
|
||||||
|
return "\n".join(unfolded)
|
||||||
|
|
||||||
|
|
||||||
|
def get_calendar_name(raw_ical: str) -> str | None:
|
||||||
|
"""Extrait le nom du calendrier depuis la propriété ``X-WR-CALNAME``.
|
||||||
|
|
||||||
|
La propriété peut comporter des paramètres (par exemple
|
||||||
|
``X-WR-CALNAME;LANGUAGE=fr:Nom``) et les lignes de continuation
|
||||||
|
iCalendar sont dépliées avant la recherche.
|
||||||
|
|
||||||
|
:param raw_ical: Contenu brut du flux iCal.
|
||||||
|
:return: Nom du calendrier ou ``None`` si la propriété est absente.
|
||||||
|
:rtype: str | None
|
||||||
|
"""
|
||||||
|
match = _CALNAME_PATTERN.search(_unfold_ical(raw_ical))
|
||||||
|
if match is None:
|
||||||
|
return None
|
||||||
|
return match.group(1).strip()
|
||||||
|
|
||||||
|
|
||||||
|
def split_header_and_body(description: str) -> tuple[str, str]:
|
||||||
|
"""Sépare l'en-tête texte du corps HTML dans la DESCRIPTION.
|
||||||
|
|
||||||
|
L'en-tête est la partie avant le premier ``<strong>``, le corps
|
||||||
|
HTML commence à partir du premier ``<strong>``.
|
||||||
|
|
||||||
|
:param description: Contenu brut de la DESCRIPTION d'un VEVENT.
|
||||||
|
:return: Tuple ``(en-tête, corps)`` ; le corps est vide si aucun ``<strong>``.
|
||||||
|
:rtype: tuple[str, str]
|
||||||
|
"""
|
||||||
|
strong_start = description.find("<strong>")
|
||||||
|
if strong_start == -1:
|
||||||
|
return description.strip(), ""
|
||||||
|
header = description[:strong_start].strip()
|
||||||
|
body = description[strong_start:]
|
||||||
|
return header, body
|
||||||
|
|
||||||
|
|
||||||
|
def parse_header(header: str) -> HeaderInfo:
|
||||||
|
"""Parse l'en-tête texte pour extraire les métadonnées du cours.
|
||||||
|
|
||||||
|
Les labels reconnus sont : ``Matière :``, ``Professeur(s) :``,
|
||||||
|
``Salle(s) :`` et ``Groupe :``. La recherche se fait par position
|
||||||
|
des labels, ce qui supporte aussi bien un en-tête multi-lignes
|
||||||
|
qu'un en-tête dont les lignes sont jointes sur une seule ligne.
|
||||||
|
|
||||||
|
:param header: En-tête texte (avant le premier ``<strong>``).
|
||||||
|
:return: Dictionnaire typé avec les champs subject, teachers, rooms, group.
|
||||||
|
:rtype: HeaderInfo
|
||||||
|
"""
|
||||||
|
info: HeaderInfo = {"subject": "", "teachers": [], "rooms": [], "group": None}
|
||||||
|
matches = list(_HEADER_LABEL_PATTERN.finditer(header))
|
||||||
|
for index, match in enumerate(matches):
|
||||||
|
value_start = match.end()
|
||||||
|
value_end = matches[index + 1].start() if index + 1 < len(matches) else len(header)
|
||||||
|
value = unescape(header[value_start:value_end].strip())
|
||||||
|
label = match.group(1).lower()
|
||||||
|
if label == "matière":
|
||||||
|
info["subject"] = value
|
||||||
|
elif label in ("professeur", "professeurs"):
|
||||||
|
info["teachers"] = [part.strip() for part in value.split(",") if part.strip()]
|
||||||
|
elif label in ("salle", "salles"):
|
||||||
|
info["rooms"] = [part.strip() for part in value.split(",") if part.strip()]
|
||||||
|
elif label == "groupe":
|
||||||
|
info["group"] = value
|
||||||
|
return info
|
||||||
|
|
||||||
|
|
||||||
|
def _strip_html(text: str) -> str:
|
||||||
|
"""Retire les balises HTML d'un texte et nettoie les espaces.
|
||||||
|
|
||||||
|
:param text: Texte pouvant contenir des balises HTML.
|
||||||
|
:return: Texte brut sans balises, entités HTML décodées.
|
||||||
|
:rtype: str
|
||||||
|
"""
|
||||||
|
cleaned = _TAG_PATTERN.sub("", text)
|
||||||
|
return unescape(cleaned).strip()
|
||||||
|
|
||||||
|
|
||||||
|
def _parse_french_date(value: str) -> date | None:
|
||||||
|
"""Convertit une date au format ``JJ/MM/AAAA`` en :class:`date`.
|
||||||
|
|
||||||
|
:param value: Date au format ``JJ/MM/AAAA``.
|
||||||
|
:return: Date parsée, ou ``None`` si le format est invalide.
|
||||||
|
:rtype: date | None
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
return datetime.strptime(value, "%d/%m/%Y").date()
|
||||||
|
except ValueError:
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def parse_body(body: str) -> tuple[str | None, list[tuple[date, str]], list[tuple[date, str]]]:
|
||||||
|
"""Parse le corps HTML pour extraire contenu pédagogique et devoirs.
|
||||||
|
|
||||||
|
Le contenu est extrait de la section ``<strong>Contenu pédagogique :</strong>``.
|
||||||
|
Les devoirs à faire sont extraits des sections ``<strong>Pour le JJ/MM/AAAA :</strong>``
|
||||||
|
(liste de tuples ``(date, texte)`` dans l'ordre du flux) et les devoirs donnés
|
||||||
|
des sections ``<strong>Donné le JJ/MM/AAAA :</strong>`` (liste de tuples
|
||||||
|
``(date, texte)``). Les listes préservent tous les blocs, même lorsque plusieurs
|
||||||
|
sections partagent la même date.
|
||||||
|
|
||||||
|
:param body: Corps HTML (à partir du premier ``<strong>``).
|
||||||
|
:return: Tuple ``(contenu pédagogique, devoirs dus, devoirs donnés)``.
|
||||||
|
:rtype: tuple[str | None, list[tuple[date, str]], list[tuple[date, str]]]
|
||||||
|
"""
|
||||||
|
content: str | None = None
|
||||||
|
due_blocks: list[tuple[date, str]] = []
|
||||||
|
assigned_blocks: list[tuple[date, str]] = []
|
||||||
|
|
||||||
|
content_match = _CONTENT_PATTERN.search(body)
|
||||||
|
if content_match is not None:
|
||||||
|
content = _strip_html(content_match.group(1))
|
||||||
|
|
||||||
|
for match in _DUE_PATTERN.finditer(body):
|
||||||
|
due_date = _parse_french_date(match.group(1))
|
||||||
|
if due_date is not None:
|
||||||
|
due_blocks.append((due_date, _strip_html(match.group(2))))
|
||||||
|
|
||||||
|
for match in _ASSIGNED_PATTERN.finditer(body):
|
||||||
|
assigned_date = _parse_french_date(match.group(1))
|
||||||
|
if assigned_date is not None:
|
||||||
|
assigned_blocks.append((assigned_date, _strip_html(match.group(2))))
|
||||||
|
|
||||||
|
return content, due_blocks, assigned_blocks
|
||||||
|
|
||||||
|
|
||||||
|
def parse_homework_blocks(
|
||||||
|
due_blocks: list[tuple[date, str]],
|
||||||
|
assigned_blocks: list[tuple[date, str]],
|
||||||
|
) -> tuple[HomeworkBlock, ...]:
|
||||||
|
"""Construit les :class:`HomeworkBlock` depuis les listes de devoirs.
|
||||||
|
|
||||||
|
Les blocs dus (``kind="due"``) précèdent les blocs donnés
|
||||||
|
(``kind="assigned"``), dans l'ordre des listes. Tous les blocs
|
||||||
|
sont préservés, y compris lorsque plusieurs partagent la même date.
|
||||||
|
|
||||||
|
:param due_blocks: Liste de tuples ``(date, texte)`` des devoirs à faire.
|
||||||
|
:param assigned_blocks: Liste de tuples ``(date, texte)`` des devoirs donnés.
|
||||||
|
:return: Tuple de blocs de devoirs pour le cours.
|
||||||
|
:rtype: tuple[HomeworkBlock, ...]
|
||||||
|
"""
|
||||||
|
blocks: list[HomeworkBlock] = []
|
||||||
|
for due_date, text in due_blocks:
|
||||||
|
blocks.append(HomeworkBlock(kind="due", date=due_date, text=text, html=text))
|
||||||
|
for assigned_date, text in assigned_blocks:
|
||||||
|
blocks.append(HomeworkBlock(kind="assigned", date=assigned_date, text=text, html=text))
|
||||||
|
return tuple(blocks)
|
||||||
|
|
||||||
|
|
||||||
|
def normalize_homework_text(text: str) -> str:
|
||||||
|
"""Normalise le texte d'un devoir pour la déduplication.
|
||||||
|
|
||||||
|
Unifie les espaces multiples, supprime les balises HTML, puis
|
||||||
|
applique trim et minuscules.
|
||||||
|
|
||||||
|
:param text: Texte brut du devoir.
|
||||||
|
:return: Texte normalisé.
|
||||||
|
:rtype: str
|
||||||
|
"""
|
||||||
|
normalized = re.sub(r"\s+", " ", text)
|
||||||
|
normalized = _TAG_PATTERN.sub("", normalized)
|
||||||
|
return normalized.strip().lower()
|
||||||
|
|
||||||
|
|
||||||
|
def generate_homework_id(due_on: date, normalized_text: str) -> str:
|
||||||
|
"""Génère un ID stable pour un devoir.
|
||||||
|
|
||||||
|
L'ID est la clé ``AAAA-MM-JJ|texte_normalisé`` hachée en SHA-1 dont
|
||||||
|
on garde les 12 premiers caractères hexadécimaux. Le hachage n'est
|
||||||
|
pas utilisé à des fins de sécurité (``usedforsecurity=False``).
|
||||||
|
|
||||||
|
:param due_on: Date d'échéance du devoir.
|
||||||
|
:param normalized_text: Texte normalisé du devoir.
|
||||||
|
:return: ID stable (12 caractères hexadécimaux).
|
||||||
|
:rtype: str
|
||||||
|
"""
|
||||||
|
payload = f"{due_on.isoformat()}|{normalized_text}".encode()
|
||||||
|
return hashlib.sha1(payload, usedforsecurity=False).hexdigest()[:12]
|
||||||
|
|
||||||
|
|
||||||
|
def collect_homeworks(lessons: list[Lesson], target_date: date) -> list[Homework]:
|
||||||
|
"""Collecte et déduplique les devoirs en deux passes globales.
|
||||||
|
|
||||||
|
Passe 1 : les blocs ``due`` (devoirs à faire pour ``target_date``)
|
||||||
|
de tous les cours. Passe 2 : les blocs ``assigned`` (devoirs donnés
|
||||||
|
le jour cible) des cours du jour ``target_date``. La déduplication
|
||||||
|
se fait par texte normalisé (premier venu, premier servi) et le
|
||||||
|
résultat est trié par matière puis texte.
|
||||||
|
|
||||||
|
:param lessons: Liste de tous les cours (VEVENT) parsés.
|
||||||
|
:param target_date: Date cible pour laquelle collecter les devoirs.
|
||||||
|
:return: Liste unique de devoirs, triée par matière puis texte.
|
||||||
|
:rtype: list[Homework]
|
||||||
|
"""
|
||||||
|
by_text: dict[str, Homework] = {}
|
||||||
|
|
||||||
|
for lesson in lessons:
|
||||||
|
for block in lesson.homework_blocks:
|
||||||
|
if block.kind == "due" and block.date == target_date:
|
||||||
|
key = normalize_homework_text(block.text)
|
||||||
|
if key not in by_text:
|
||||||
|
by_text[key] = Homework(
|
||||||
|
id=generate_homework_id(target_date, key),
|
||||||
|
subject=lesson.subject,
|
||||||
|
teachers=lesson.teachers,
|
||||||
|
assigned_on=lesson.start.date(),
|
||||||
|
due_on=block.date,
|
||||||
|
text=block.text,
|
||||||
|
html=block.html,
|
||||||
|
)
|
||||||
|
|
||||||
|
for lesson in lessons:
|
||||||
|
if lesson.start.date() != target_date:
|
||||||
|
continue
|
||||||
|
for block in lesson.homework_blocks:
|
||||||
|
if block.kind == "assigned":
|
||||||
|
key = normalize_homework_text(block.text)
|
||||||
|
if key not in by_text:
|
||||||
|
by_text[key] = Homework(
|
||||||
|
id=generate_homework_id(target_date, key),
|
||||||
|
subject=lesson.subject,
|
||||||
|
teachers=lesson.teachers,
|
||||||
|
assigned_on=block.date,
|
||||||
|
due_on=target_date,
|
||||||
|
text=block.text,
|
||||||
|
html=block.html,
|
||||||
|
)
|
||||||
|
|
||||||
|
return sorted(by_text.values(), key=lambda hw: (hw.subject.lower(), hw.text.lower()))
|
||||||
|
|
||||||
|
|
||||||
|
def parse_ical(raw_ical: str) -> tuple[list[Lesson], list[Homework], list[SchoolEvent]]:
|
||||||
|
"""Parse un flux iCal Pronote en événements typés.
|
||||||
|
|
||||||
|
Les VEVENT de vacances/congés (tout le jour) deviennent des
|
||||||
|
:class:`SchoolEvent` de type ``holiday``. Les VEVENT horodatés
|
||||||
|
deviennent des :class:`Lesson` dont le statut dérive de la propriété
|
||||||
|
``STATUS`` (``CANCELLED`` → ``CANCELLED``) et de la catégorie
|
||||||
|
(``Cours - Cours annulé`` → ``CANCELLED``, ``Cours - Cours déplacé`` →
|
||||||
|
``MOVED``). Les UID sont normalisés ;
|
||||||
|
un événement sans UID reçoit un UID déterministe généré à partir
|
||||||
|
de ses champs clés (début, fin, matière, enseignants, salles, groupe).
|
||||||
|
|
||||||
|
:param raw_ical: Contenu brut du flux iCal.
|
||||||
|
:return: Tuple ``(lessons, homeworks, school_events)`` où ``homeworks``
|
||||||
|
est **toujours vide** : la collecte/déduplication se fait plus tard
|
||||||
|
via :func:`collect_homeworks` une fois la date cible connue.
|
||||||
|
:rtype: tuple[list[Lesson], list[Homework], list[SchoolEvent]]
|
||||||
|
"""
|
||||||
|
calendar = Calendar.from_ical(raw_ical)
|
||||||
|
|
||||||
|
lessons: list[Lesson] = []
|
||||||
|
homeworks: list[Homework] = []
|
||||||
|
school_events: list[SchoolEvent] = []
|
||||||
|
|
||||||
|
for component in calendar.walk():
|
||||||
|
if component.name != "VEVENT":
|
||||||
|
continue
|
||||||
|
|
||||||
|
dtstart = component.get("dtstart")
|
||||||
|
dtend = component.get("dtend")
|
||||||
|
if dtstart is None or dtend is None:
|
||||||
|
continue
|
||||||
|
start = dtstart.dt
|
||||||
|
end = dtend.dt
|
||||||
|
|
||||||
|
categories_obj = component.get("categories")
|
||||||
|
if categories_obj is None:
|
||||||
|
categories: list[str] = []
|
||||||
|
else:
|
||||||
|
categories = [str(category) for category in categories_obj.cats]
|
||||||
|
|
||||||
|
# Événements de type vacances/congés (tout le jour).
|
||||||
|
if any(cat in ("Congés", "Vacances") for cat in categories):
|
||||||
|
from_date = start.date() if isinstance(start, datetime) else start
|
||||||
|
to_date = end.date() if isinstance(end, datetime) else end
|
||||||
|
summary = component.get("summary")
|
||||||
|
label = str(summary) if summary is not None else ""
|
||||||
|
school_events.append(
|
||||||
|
SchoolEvent(
|
||||||
|
kind=SchoolEventKind.HOLIDAY,
|
||||||
|
label=label,
|
||||||
|
from_date=from_date,
|
||||||
|
to_date=to_date,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
|
||||||
|
# Les cours sont des événements horodatés ; les événements tout
|
||||||
|
# le jour non scolaires sont ignorés.
|
||||||
|
if not isinstance(start, datetime) or not isinstance(end, datetime):
|
||||||
|
continue
|
||||||
|
|
||||||
|
status_obj = component.get("status")
|
||||||
|
status_value = str(status_obj).strip().upper() if status_obj is not None else ""
|
||||||
|
if status_value == "CANCELLED" or "Cours - Cours annulé" in categories:
|
||||||
|
status = LessonStatus.CANCELLED
|
||||||
|
elif "Cours - Cours déplacé" in categories:
|
||||||
|
status = LessonStatus.MOVED
|
||||||
|
else:
|
||||||
|
status = LessonStatus.NORMAL
|
||||||
|
|
||||||
|
description = component.get("description")
|
||||||
|
description_str = str(description) if description is not None else ""
|
||||||
|
header, body = split_header_and_body(description_str)
|
||||||
|
lesson_data = parse_header(header)
|
||||||
|
content, due_blocks, assigned_blocks = parse_body(body)
|
||||||
|
homework_blocks = parse_homework_blocks(due_blocks, assigned_blocks)
|
||||||
|
|
||||||
|
uid_value = component.get("uid")
|
||||||
|
if uid_value is None or str(uid_value).strip() == "":
|
||||||
|
uid = generate_deterministic_uid(
|
||||||
|
start=start,
|
||||||
|
end=end,
|
||||||
|
subject=lesson_data["subject"],
|
||||||
|
teachers=lesson_data["teachers"],
|
||||||
|
rooms=lesson_data["rooms"],
|
||||||
|
group=lesson_data["group"],
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
uid = normalize_pronote_uid(str(uid_value))
|
||||||
|
|
||||||
|
lessons.append(
|
||||||
|
Lesson(
|
||||||
|
id=uid,
|
||||||
|
start=start,
|
||||||
|
end=end,
|
||||||
|
subject=lesson_data["subject"],
|
||||||
|
teachers=tuple(lesson_data["teachers"]),
|
||||||
|
rooms=tuple(lesson_data["rooms"]),
|
||||||
|
group=lesson_data["group"],
|
||||||
|
status=status,
|
||||||
|
content=content,
|
||||||
|
homework_blocks=homework_blocks,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
return lessons, homeworks, school_events
|
||||||
57
tests/fixtures/pronote-4e.ics
vendored
Normal file
57
tests/fixtures/pronote-4e.ics
vendored
Normal file
@@ -0,0 +1,57 @@
|
|||||||
|
BEGIN:VCALENDAR
|
||||||
|
VERSION:2.0
|
||||||
|
PRODID:-//Index Education//Pronote//FR
|
||||||
|
X-WR-CALNAME:Classe de 4e
|
||||||
|
BEGIN:VEVENT
|
||||||
|
UID:Edt_12345@index-education.net-20260905T120000Z-Index-Education
|
||||||
|
DTSTAMP:20260905T120000Z
|
||||||
|
DTSTART:20260905T080000Z
|
||||||
|
DTEND:20260905T090000Z
|
||||||
|
SUMMARY:Mathématiques
|
||||||
|
CATEGORIES:Cours
|
||||||
|
DESCRIPTION:<div>
|
||||||
|
Matière : Mathématiques
|
||||||
|
Professeur : M. Dupont
|
||||||
|
Salle : 204
|
||||||
|
Groupe : Classe entière
|
||||||
|
|
||||||
|
<strong>Contenu pédagogique :
|
||||||
|
</strong>
|
||||||
|
Résoudre des équations du second degré.
|
||||||
|
<strong>Pour le 10/09/2026 :
|
||||||
|
</strong>
|
||||||
|
Exercices 1 à 5 page 42.
|
||||||
|
<strong>Donné le 05/09/2026 :
|
||||||
|
</strong>
|
||||||
|
Exercices 1 à 5 page 42.
|
||||||
|
</div>
|
||||||
|
END:VEVENT
|
||||||
|
BEGIN:VEVENT
|
||||||
|
UID:Edt_67890@index-education.net-20260905T120000Z-Index-Education
|
||||||
|
DTSTAMP:20260905T120000Z
|
||||||
|
DTSTART:20260905T090000Z
|
||||||
|
DTEND:20260905T100000Z
|
||||||
|
SUMMARY:Français
|
||||||
|
CATEGORIES:Cours - Cours annulé
|
||||||
|
STATUS:CANCELLED
|
||||||
|
DESCRIPTION:<div>
|
||||||
|
Matière : Français
|
||||||
|
Professeur : Mme Martin
|
||||||
|
Salle : 205
|
||||||
|
Groupe : Classe entière
|
||||||
|
|
||||||
|
<strong>Contenu pédagogique :
|
||||||
|
</strong>
|
||||||
|
Étude d'un texte littéraire.
|
||||||
|
</div>
|
||||||
|
END:VEVENT
|
||||||
|
BEGIN:VEVENT
|
||||||
|
UID:Edt_11111@index-education.net-20260905T120000Z-Index-Education
|
||||||
|
DTSTAMP:20260905T120000Z
|
||||||
|
DTSTART;VALUE=DATE:20261020
|
||||||
|
DTEND;VALUE=DATE:20261021
|
||||||
|
SUMMARY:Vacances de la Toussaint
|
||||||
|
CATEGORIES:Congés
|
||||||
|
DESCRIPTION:Période de vacances scolaires.
|
||||||
|
END:VEVENT
|
||||||
|
END:VCALENDAR
|
||||||
987
tests/unit/test_fallback.py
Normal file
987
tests/unit/test_fallback.py
Normal file
@@ -0,0 +1,987 @@
|
|||||||
|
"""Tests unitaires pour la logique de repli iCal / pronotepy.
|
||||||
|
|
||||||
|
Ce module valide le comportement du module :mod:`pronote_sync.sources.pronote.fallback`
|
||||||
|
et de son implémentation :class:`PronoteFetcher`. Les tests couvrent :
|
||||||
|
|
||||||
|
- La sélection de la source d'agenda (``ical``, ``pronotepy``, ``auto``) et de devoirs,
|
||||||
|
- Le repli automatique iCal → pronotepy en mode ``auto``,
|
||||||
|
- La levée de :class:`PipelineCriticalError` lorsque toutes les sources échouent,
|
||||||
|
- Le masquage des secrets dans les messages d'erreur,
|
||||||
|
- Les méthodes toujours basées sur pronotepy (messages, informations).
|
||||||
|
|
||||||
|
Tous les appels réseau et les interactions avec pronotepy sont mockés.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from datetime import date, datetime
|
||||||
|
from typing import TYPE_CHECKING
|
||||||
|
from unittest.mock import MagicMock, patch
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from pydantic import SecretStr
|
||||||
|
|
||||||
|
from pronote_sync.config.settings import PronoteSettings, Settings
|
||||||
|
from pronote_sync.errors import PipelineCriticalError
|
||||||
|
from pronote_sync.models.agenda import (
|
||||||
|
HomeworkBlock,
|
||||||
|
Lesson,
|
||||||
|
LessonStatus,
|
||||||
|
SchoolEvent,
|
||||||
|
SchoolEventKind,
|
||||||
|
)
|
||||||
|
from pronote_sync.models.homework import Homework
|
||||||
|
from pronote_sync.models.message import Message, MessageType
|
||||||
|
from pronote_sync.sources.pronote.fallback import PronoteFetcher
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from typing import Protocol
|
||||||
|
|
||||||
|
class _MockPronoteClientProtocol(Protocol):
|
||||||
|
def get_messages(self) -> list[Message]: ...
|
||||||
|
def get_informations(self) -> list[Message]: ...
|
||||||
|
def get_lessons(self, start: date, end: date) -> list[Lesson]: ...
|
||||||
|
def get_homeworks(self, start: date, end: date) -> list[Homework]: ...
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(name="mock_settings")
|
||||||
|
def fixture_mock_settings() -> Settings:
|
||||||
|
"""Fixture fournissant une configuration Settings adaptée aux tests.
|
||||||
|
|
||||||
|
:return: Instance de :class:`Settings` avec des valeurs sûres pour les tests.
|
||||||
|
:rtype: Settings
|
||||||
|
"""
|
||||||
|
return Settings(
|
||||||
|
pronote=PronoteSettings(
|
||||||
|
pronote_url="https://pronote.example.com",
|
||||||
|
ical_url=SecretStr("file:///fake/ical.ics"),
|
||||||
|
agenda_source="auto",
|
||||||
|
homework_source="auto",
|
||||||
|
username="testuser",
|
||||||
|
password=SecretStr("testpass"),
|
||||||
|
ent="ent",
|
||||||
|
),
|
||||||
|
app=Settings().app,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(name="mock_fetcher")
|
||||||
|
def fixture_mock_fetcher(mock_settings: Settings) -> PronoteFetcher:
|
||||||
|
"""Fixture fournissant une instance de :class:`PronoteFetcher` prête à l'emploi.
|
||||||
|
|
||||||
|
:param mock_settings: Configuration de test.
|
||||||
|
:return: Instance de :class:`PronoteFetcher` pour les tests.
|
||||||
|
:rtype: PronoteFetcher
|
||||||
|
"""
|
||||||
|
client: _MockPronoteClientProtocol = MagicMock()
|
||||||
|
return PronoteFetcher(settings=mock_settings, pronote_client=client)
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_agenda_ical_mode(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test la récupération de l'agenda en mode source iCal.
|
||||||
|
|
||||||
|
On mock ``fetch_ical`` et ``parse_ical`` pour retourner des cours et événements.
|
||||||
|
On vérifie que le fetcher retourne bien ces données.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
start_dt = datetime(2025, 9, 1, 8, 0)
|
||||||
|
end_dt = datetime(2025, 9, 1, 9, 30)
|
||||||
|
lessons = [
|
||||||
|
Lesson(
|
||||||
|
id="l1",
|
||||||
|
start=start_dt,
|
||||||
|
end=end_dt,
|
||||||
|
subject="Maths",
|
||||||
|
teachers=("Dupont",),
|
||||||
|
rooms=("S1",),
|
||||||
|
group="2ndeA",
|
||||||
|
status=LessonStatus.NORMAL,
|
||||||
|
content=None,
|
||||||
|
)
|
||||||
|
]
|
||||||
|
events = [
|
||||||
|
SchoolEvent(
|
||||||
|
kind=SchoolEventKind.HOLIDAY,
|
||||||
|
label="Vacances",
|
||||||
|
from_date=date(2025, 9, 1),
|
||||||
|
to_date=date(2025, 9, 15),
|
||||||
|
)
|
||||||
|
]
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
|
):
|
||||||
|
m_fetch_ical.return_value = "BEGIN:VCALENDAR\n..."
|
||||||
|
m_parse_ical.return_value = (lessons, [], events)
|
||||||
|
|
||||||
|
result_lessons, result_events = mock_fetcher.fetch_agenda()
|
||||||
|
|
||||||
|
assert result_lessons == lessons
|
||||||
|
assert result_events == events
|
||||||
|
m_fetch_ical.assert_called_once()
|
||||||
|
m_parse_ical.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_agenda_pronotepy_mode(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test la récupération de l'agenda en mode source pronotepy.
|
||||||
|
|
||||||
|
On mock ``get_lessons`` du client pour retourner des cours.
|
||||||
|
On vérifie que le fetcher retourne ces cours (événements scolaires vides).
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
start_dt = datetime(2025, 9, 1, 8, 0)
|
||||||
|
end_dt = datetime(2025, 9, 1, 9, 30)
|
||||||
|
lessons = [
|
||||||
|
Lesson(
|
||||||
|
id="l1",
|
||||||
|
start=start_dt,
|
||||||
|
end=end_dt,
|
||||||
|
subject="Physique",
|
||||||
|
teachers=("Martin",),
|
||||||
|
rooms=("Labo1",),
|
||||||
|
group="1ereB",
|
||||||
|
status=LessonStatus.NORMAL,
|
||||||
|
content=None,
|
||||||
|
)
|
||||||
|
]
|
||||||
|
|
||||||
|
client = MagicMock()
|
||||||
|
client.get_lessons.return_value = lessons
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
|
||||||
|
result_lessons, result_events = mock_fetcher.fetch_agenda()
|
||||||
|
|
||||||
|
assert result_lessons == lessons
|
||||||
|
assert result_events == []
|
||||||
|
client.get_lessons.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_agenda_auto_ical_success(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test le mode auto : succès de l'iCal, pronotepy non appelé.
|
||||||
|
|
||||||
|
On mock iCal pour réussir, et on vérifie que pronotepy n'est pas sollicité.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
start_dt = datetime(2025, 9, 1, 8, 0)
|
||||||
|
end_dt = datetime(2025, 9, 1, 9, 30)
|
||||||
|
lessons = [
|
||||||
|
Lesson(
|
||||||
|
id="l1",
|
||||||
|
start=start_dt,
|
||||||
|
end=end_dt,
|
||||||
|
subject="SVT",
|
||||||
|
teachers=("Durand",),
|
||||||
|
rooms=("S2",),
|
||||||
|
group="3emeC",
|
||||||
|
status=LessonStatus.NORMAL,
|
||||||
|
content=None,
|
||||||
|
)
|
||||||
|
]
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
|
):
|
||||||
|
m_fetch_ical.return_value = "BEGIN:VCALENDAR\n..."
|
||||||
|
m_parse_ical.return_value = (lessons, [], [])
|
||||||
|
|
||||||
|
result_lessons, _ = mock_fetcher.fetch_agenda()
|
||||||
|
|
||||||
|
assert result_lessons == lessons
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_agenda_auto_fallback_to_pronotepy(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test le mode auto : échec iCal, repli sur pronotepy.
|
||||||
|
|
||||||
|
On mock iCal pour échouer, pronotepy pour réussir. On vérifie que pronotepy est appelé.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
start_dt = datetime(2025, 9, 1, 8, 0)
|
||||||
|
end_dt = datetime(2025, 9, 1, 9, 30)
|
||||||
|
lessons = [
|
||||||
|
Lesson(
|
||||||
|
id="l1",
|
||||||
|
start=start_dt,
|
||||||
|
end=end_dt,
|
||||||
|
subject="Histoire",
|
||||||
|
teachers=("Lefevre",),
|
||||||
|
rooms=("S3",),
|
||||||
|
group="2ndeD",
|
||||||
|
status=LessonStatus.NORMAL,
|
||||||
|
content=None,
|
||||||
|
)
|
||||||
|
]
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
|
):
|
||||||
|
m_fetch_ical.side_effect = OSError("iCal unreachable")
|
||||||
|
m_parse_ical.side_effect = OSError("iCal parse error")
|
||||||
|
client = MagicMock()
|
||||||
|
client.get_lessons.return_value = lessons
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
|
||||||
|
result_lessons, _ = mock_fetcher.fetch_agenda()
|
||||||
|
|
||||||
|
assert result_lessons == lessons
|
||||||
|
client.get_lessons.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_agenda_auto_both_fail(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test le mode auto : échec des deux sources → PipelineCriticalError.
|
||||||
|
|
||||||
|
On mock iCal et pronotepy pour échouer. On vérifie la levée de l'erreur critique.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
# Disable pronotepy so fallback is None
|
||||||
|
mock_fetcher._settings.pronote.pronote_url = None
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
|
):
|
||||||
|
m_fetch_ical.side_effect = OSError("iCal unreachable")
|
||||||
|
m_parse_ical.side_effect = OSError("iCal parse error")
|
||||||
|
client = MagicMock()
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
|
||||||
|
with pytest.raises(PipelineCriticalError) as exc_info:
|
||||||
|
mock_fetcher.fetch_agenda()
|
||||||
|
|
||||||
|
assert "la source ical a échoué" in str(exc_info.value)
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_agenda_ical_mode_failure(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test le mode ical : échec → PipelineCriticalError masquée.
|
||||||
|
|
||||||
|
On mock iCal pour échouer et désactive pronotepy pour que le repli soit None.
|
||||||
|
On vérifie que l'erreur brute est masquée dans la levée.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
# Override settings to use ical mode explicitly and disable fallback
|
||||||
|
mock_fetcher._settings.pronote.agenda_source = "ical"
|
||||||
|
mock_fetcher._settings.pronote.pronote_url = None
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
|
):
|
||||||
|
m_fetch_ical.side_effect = OSError(
|
||||||
|
"Impossible de lire le fichier iCal file:///fake/ical.ics : iCal unreachable"
|
||||||
|
)
|
||||||
|
m_parse_ical.side_effect = OSError("iCal parse error")
|
||||||
|
|
||||||
|
with pytest.raises(PipelineCriticalError) as exc_info:
|
||||||
|
mock_fetcher.fetch_agenda()
|
||||||
|
|
||||||
|
assert "Impossible de récupérer l'agenda : la source ical a échoué" in str(exc_info.value)
|
||||||
|
# Vérifie que le message ne contient pas de secret
|
||||||
|
assert "file:///fake/ical.ics" not in str(exc_info.value)
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_agenda_pronotepy_mode_failure(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test le mode pronotepy : échec → PipelineCriticalError masquée.
|
||||||
|
|
||||||
|
On mock pronotepy pour échouer et désactive iCal pour que le repli soit None.
|
||||||
|
On vérifie que l'erreur est masquée dans la levée.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
# Override settings to use pronotepy mode explicitly and disable fallback
|
||||||
|
mock_fetcher._settings.pronote.agenda_source = "pronotepy"
|
||||||
|
mock_fetcher._settings.pronote.ical_url = None
|
||||||
|
|
||||||
|
client = MagicMock()
|
||||||
|
client.get_lessons.side_effect = OSError("Pronote API error")
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
|
||||||
|
with pytest.raises(PipelineCriticalError) as exc_info:
|
||||||
|
mock_fetcher.fetch_agenda()
|
||||||
|
|
||||||
|
assert "Impossible de récupérer l'agenda : la source pronotepy a échoué" in str(exc_info.value)
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_homework_ical_mode(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test la récupération des devoirs en mode source iCal.
|
||||||
|
|
||||||
|
On mock iCal pour retourner des cours avec blocs de devoirs, et on cible une date.
|
||||||
|
On vérifie que les devoirs sont correctement collectés.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
target_date = date(2025, 9, 5)
|
||||||
|
|
||||||
|
start_dt = datetime(2025, 9, 1, 8, 0)
|
||||||
|
end_dt = datetime(2025, 9, 1, 9, 30)
|
||||||
|
lessons = [
|
||||||
|
Lesson(
|
||||||
|
id="l1",
|
||||||
|
start=start_dt,
|
||||||
|
end=end_dt,
|
||||||
|
subject="Maths",
|
||||||
|
teachers=("Dupont",),
|
||||||
|
rooms=("S1",),
|
||||||
|
group="2ndeA",
|
||||||
|
status=LessonStatus.NORMAL,
|
||||||
|
content=None,
|
||||||
|
homework_blocks=(
|
||||||
|
HomeworkBlock(
|
||||||
|
kind="due",
|
||||||
|
date=target_date,
|
||||||
|
text="Devoir sur les fonctions",
|
||||||
|
html="<p>Devoir sur les fonctions</p>",
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
]
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.collect_homeworks") as m_collect,
|
||||||
|
):
|
||||||
|
m_fetch_ical.return_value = "BEGIN:VCALENDAR\n..."
|
||||||
|
m_parse_ical.return_value = (lessons, [], [])
|
||||||
|
m_collect.return_value = [
|
||||||
|
Homework(
|
||||||
|
id="hw1",
|
||||||
|
subject="Maths",
|
||||||
|
teachers=(),
|
||||||
|
assigned_on=None,
|
||||||
|
due_on=target_date,
|
||||||
|
text="Devoir sur les fonctions",
|
||||||
|
html="<p>Devoir sur les fonctions</p>",
|
||||||
|
)
|
||||||
|
]
|
||||||
|
|
||||||
|
result = mock_fetcher.fetch_homework(target_date)
|
||||||
|
|
||||||
|
assert len(result) == 1
|
||||||
|
assert result[0].subject == "Maths"
|
||||||
|
assert result[0].due_on == target_date
|
||||||
|
m_collect.assert_called_once_with(lessons, target_date)
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_homework_auto_fallback(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test le mode auto des devoirs : échec iCal, repli pronotepy.
|
||||||
|
|
||||||
|
On mock iCal pour échouer et pronotepy pour réussir. On vérifie que pronotepy est utilisé.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
target_date = date(2025, 9, 10)
|
||||||
|
homeworks = [
|
||||||
|
Homework(
|
||||||
|
id="hw1",
|
||||||
|
subject="Physique",
|
||||||
|
teachers=(),
|
||||||
|
assigned_on=None,
|
||||||
|
due_on=target_date,
|
||||||
|
text="TP à préparer",
|
||||||
|
html="TP à préparer",
|
||||||
|
)
|
||||||
|
]
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.collect_homeworks") as m_collect,
|
||||||
|
):
|
||||||
|
m_fetch_ical.side_effect = OSError("iCal unreachable")
|
||||||
|
m_parse_ical.side_effect = OSError("iCal parse error")
|
||||||
|
client = MagicMock()
|
||||||
|
client.get_homeworks.return_value = homeworks
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
m_collect.return_value = homeworks
|
||||||
|
|
||||||
|
result = mock_fetcher.fetch_homework(target_date)
|
||||||
|
|
||||||
|
assert result == homeworks
|
||||||
|
client.get_homeworks.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_homework_auto_both_fail(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test le mode auto des devoirs : échec des deux sources → PipelineCriticalError.
|
||||||
|
|
||||||
|
On mock iCal et pronotepy pour échouer. On vérifie la levée de l'erreur critique.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
target_date = date(2025, 9, 10)
|
||||||
|
|
||||||
|
# Disable pronotepy so fallback is None
|
||||||
|
mock_fetcher._settings.pronote.pronote_url = None
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
|
):
|
||||||
|
m_fetch_ical.side_effect = OSError("iCal unreachable")
|
||||||
|
m_parse_ical.side_effect = OSError("iCal parse error")
|
||||||
|
client = MagicMock()
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
|
||||||
|
with pytest.raises(PipelineCriticalError) as exc_info:
|
||||||
|
mock_fetcher.fetch_homework(target_date)
|
||||||
|
|
||||||
|
assert "la source ical a échoué" in str(exc_info.value)
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_messages(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test la récupération des messages (toujours via pronotepy).
|
||||||
|
|
||||||
|
On mock ``get_messages`` du client pour retourner des messages typés.
|
||||||
|
On vérifie que le fetcher retourne ces messages.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
messages = [
|
||||||
|
Message(
|
||||||
|
id="m1",
|
||||||
|
type=MessageType.DISCUSSION,
|
||||||
|
title="Devoir de maths",
|
||||||
|
content="À faire pour demain",
|
||||||
|
author="M. Dupont",
|
||||||
|
date=datetime(2025, 9, 1, 10, 0),
|
||||||
|
read=False,
|
||||||
|
)
|
||||||
|
]
|
||||||
|
client = MagicMock()
|
||||||
|
client.get_messages.return_value = messages
|
||||||
|
client.get_informations.return_value = []
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
|
||||||
|
result = mock_fetcher.fetch_messages()
|
||||||
|
|
||||||
|
assert result == messages
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_informations(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test la récupération des informations (toujours via pronotepy).
|
||||||
|
|
||||||
|
On mock ``get_informations`` du client pour retourner des informations.
|
||||||
|
On vérifie que le fetcher retourne ces informations.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
infos = [
|
||||||
|
Message(
|
||||||
|
id="i1",
|
||||||
|
type=MessageType.INFORMATION,
|
||||||
|
title="Info rentrée",
|
||||||
|
content="Rappel des consignes",
|
||||||
|
author="CPE",
|
||||||
|
date=datetime(2025, 9, 1, 9, 0),
|
||||||
|
read=True,
|
||||||
|
)
|
||||||
|
]
|
||||||
|
client = MagicMock()
|
||||||
|
client.get_informations.return_value = infos
|
||||||
|
client.get_messages.return_value = []
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
|
||||||
|
result = mock_fetcher.fetch_informations()
|
||||||
|
|
||||||
|
assert result == infos
|
||||||
|
|
||||||
|
|
||||||
|
def test_no_secrets_in_error_messages(
|
||||||
|
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
|
||||||
|
) -> None:
|
||||||
|
"""Test que les messages d'erreur ne contiennent pas de secrets.
|
||||||
|
|
||||||
|
On simule une erreur iCal contenant un token ``icalsecurise`` et on vérifie que le log
|
||||||
|
masqué ne contient pas le token.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:param caplog: Fixture pytest pour capturer les logs.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
# Disable pronotepy so fallback is None to trigger PipelineCriticalError
|
||||||
|
mock_fetcher._settings.pronote.pronote_url = None
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
|
):
|
||||||
|
error_msg = (
|
||||||
|
"Impossible de lire le fichier iCal file:///ical?icalsecurise=SECRET_TOKEN_123 : "
|
||||||
|
"[Errno 2] No such file or directory"
|
||||||
|
)
|
||||||
|
m_fetch_ical.side_effect = OSError(error_msg)
|
||||||
|
m_parse_ical.side_effect = OSError("parse error")
|
||||||
|
client = MagicMock()
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
|
||||||
|
with pytest.raises(PipelineCriticalError):
|
||||||
|
mock_fetcher.fetch_agenda()
|
||||||
|
|
||||||
|
# Vérifie que le log contient la version masquée
|
||||||
|
assert "SECRET_TOKEN_123" not in caplog.text
|
||||||
|
assert "icalsecurise=REDACTED" in caplog.text or "icalsecurise" not in caplog.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_agenda_ical_mode_fallback_to_pronotepy(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test le mode ICAL : échec iCal, repli sur pronotepy.
|
||||||
|
|
||||||
|
On mock iCal pour échouer, pronotepy pour réussir. On vérifie que pronotepy est appelé.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
start_dt = datetime(2025, 9, 1, 8, 0)
|
||||||
|
end_dt = datetime(2025, 9, 1, 9, 30)
|
||||||
|
lessons = [
|
||||||
|
Lesson(
|
||||||
|
id="l1",
|
||||||
|
start=start_dt,
|
||||||
|
end=end_dt,
|
||||||
|
subject="Maths",
|
||||||
|
teachers=("Dupont",),
|
||||||
|
rooms=("S1",),
|
||||||
|
group="2ndeA",
|
||||||
|
status=LessonStatus.NORMAL,
|
||||||
|
content=None,
|
||||||
|
)
|
||||||
|
]
|
||||||
|
|
||||||
|
# Override settings to use ical mode explicitly
|
||||||
|
mock_fetcher._settings.pronote.agenda_source = "ical"
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
|
):
|
||||||
|
m_fetch_ical.side_effect = OSError("iCal unreachable")
|
||||||
|
m_parse_ical.side_effect = OSError("iCal parse error")
|
||||||
|
client = MagicMock()
|
||||||
|
client.get_lessons.return_value = lessons
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
|
||||||
|
result_lessons, result_events = mock_fetcher.fetch_agenda()
|
||||||
|
|
||||||
|
assert result_lessons == lessons
|
||||||
|
assert result_events == []
|
||||||
|
client.get_lessons.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_agenda_pronotepy_mode_fallback_to_ical(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test le mode PRONOTEPY : échec pronotepy, repli sur iCal.
|
||||||
|
|
||||||
|
On mock pronotepy pour échouer, iCal pour réussir. On vérifie que iCal est appelé.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
start_dt = datetime(2025, 9, 1, 8, 0)
|
||||||
|
end_dt = datetime(2025, 9, 1, 9, 30)
|
||||||
|
lessons = [
|
||||||
|
Lesson(
|
||||||
|
id="l1",
|
||||||
|
start=start_dt,
|
||||||
|
end=end_dt,
|
||||||
|
subject="SVT",
|
||||||
|
teachers=("Durand",),
|
||||||
|
rooms=("S2",),
|
||||||
|
group="3emeC",
|
||||||
|
status=LessonStatus.NORMAL,
|
||||||
|
content=None,
|
||||||
|
)
|
||||||
|
]
|
||||||
|
|
||||||
|
# Override settings to use pronotepy mode explicitly
|
||||||
|
mock_fetcher._settings.pronote.agenda_source = "pronotepy"
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
|
):
|
||||||
|
client = MagicMock()
|
||||||
|
client.get_lessons.side_effect = OSError("Pronote API error")
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
m_fetch_ical.return_value = "BEGIN:VCALENDAR\n..."
|
||||||
|
m_parse_ical.return_value = (lessons, [], [])
|
||||||
|
|
||||||
|
result_lessons, result_events = mock_fetcher.fetch_agenda()
|
||||||
|
|
||||||
|
assert result_lessons == lessons
|
||||||
|
assert result_events == []
|
||||||
|
m_fetch_ical.assert_called_once()
|
||||||
|
m_parse_ical.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_agenda_no_source_configured_raises(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test le mode AUTO : aucune source configurée → PipelineCriticalError.
|
||||||
|
|
||||||
|
On désactive les deux sources. On vérifie que l'erreur critique est levée.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
# Disable both sources
|
||||||
|
mock_fetcher._settings.pronote.ical_url = None
|
||||||
|
mock_fetcher._settings.pronote.pronote_url = None
|
||||||
|
|
||||||
|
with pytest.raises(PipelineCriticalError) as exc_info:
|
||||||
|
mock_fetcher.fetch_agenda()
|
||||||
|
|
||||||
|
assert "ni la source iCal ni pronotepy n'est configurée" in str(exc_info.value)
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_homework_filters_by_target_date(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test que les devoirs sont filtrés par date cible en mode pronotepy.
|
||||||
|
|
||||||
|
On mock pronotepy pour retourner des devoirs avec différentes dates d'échéance.
|
||||||
|
On vérifie que seul le devoir correspondant à la date cible est retourné.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
target_date = date(2025, 9, 10)
|
||||||
|
other_date = date(2025, 9, 11)
|
||||||
|
|
||||||
|
homeworks = [
|
||||||
|
Homework(
|
||||||
|
id="hw1",
|
||||||
|
subject="Maths",
|
||||||
|
teachers=(),
|
||||||
|
assigned_on=None,
|
||||||
|
due_on=target_date,
|
||||||
|
text="Devoir pour aujourd'hui",
|
||||||
|
html="<p>Devoir pour aujourd'hui</p>",
|
||||||
|
),
|
||||||
|
Homework(
|
||||||
|
id="hw2",
|
||||||
|
subject="Physique",
|
||||||
|
teachers=(),
|
||||||
|
assigned_on=None,
|
||||||
|
due_on=other_date,
|
||||||
|
text="Devoir pour demain",
|
||||||
|
html="<p>Devoir pour demain</p>",
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
client = MagicMock()
|
||||||
|
client.get_homeworks.return_value = homeworks
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
|
||||||
|
result = mock_fetcher.fetch_homework(target_date)
|
||||||
|
|
||||||
|
assert len(result) == 1
|
||||||
|
assert result[0].id == "hw1"
|
||||||
|
assert result[0].due_on == target_date
|
||||||
|
client.get_homeworks.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_messages_logs_and_re_raises(
|
||||||
|
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
|
||||||
|
) -> None:
|
||||||
|
"""Test que fetch_messages journalise et relance les exceptions.
|
||||||
|
|
||||||
|
On mock get_messages pour lever une exception contenant une URL secrète.
|
||||||
|
On vérifie que l'exception est relancée et que le log masqué ne contient pas l'URL.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:param caplog: Fixture pytest pour capturer les logs.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
client = MagicMock()
|
||||||
|
error_msg = "Erreur Pronote : impossible de récupérer les messages https://pronote.example.com/messages?token=SECRET_TOKEN_456"
|
||||||
|
client.get_messages.side_effect = OSError(error_msg)
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
|
||||||
|
with pytest.raises(OSError) as exc_info:
|
||||||
|
mock_fetcher.fetch_messages()
|
||||||
|
|
||||||
|
assert exc_info.value is client.get_messages.side_effect
|
||||||
|
assert "SECRET_TOKEN_456" not in caplog.text
|
||||||
|
assert (
|
||||||
|
"pronote.example.com/messages?token=REDACTED" in caplog.text
|
||||||
|
or "pronote.example.com/messages" in caplog.text
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_agenda_ical_url_none_raises_value_error(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test que _fetch_agenda_ical lève ValueError si ical_url est None.
|
||||||
|
|
||||||
|
On désactive ical_url et on appelle _fetch_agenda_ical(). On vérifie que
|
||||||
|
l'erreur ValueError est levée.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_fetcher._settings.pronote.ical_url = None
|
||||||
|
with pytest.raises(ValueError, match="PRONOTE_ICAL_URL est requis pour la source iCal"):
|
||||||
|
mock_fetcher._fetch_agenda_ical()
|
||||||
|
|
||||||
|
|
||||||
|
def test_agenda_sources_auto_only_pronotepy_configured(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test _agenda_sources en mode AUTO avec seulement pronotepy configuré.
|
||||||
|
|
||||||
|
On désactive ical_url mais on garde pronotepy configuré. On vérifie que la
|
||||||
|
source primaire est pronotepy et qu'il n'y a pas de repli.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_fetcher._settings.pronote.agenda_source = "auto"
|
||||||
|
mock_fetcher._settings.pronote.ical_url = None
|
||||||
|
primary, fallback = mock_fetcher._agenda_sources()
|
||||||
|
assert primary == "pronotepy"
|
||||||
|
assert fallback is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_agenda_fallback_both_fail_raises_pipeline_critical_error(
|
||||||
|
mock_fetcher: PronoteFetcher,
|
||||||
|
) -> None:
|
||||||
|
"""Test que fetch_agenda lève PipelineCriticalError si ICAL échoue et pronotepy aussi.
|
||||||
|
|
||||||
|
On mock ICAL pour échouer, on garde pronotepy configuré mais on mock son échec.
|
||||||
|
On vérifie que l'erreur PipelineCriticalError est levée.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_fetcher._settings.pronote.agenda_source = "ical"
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
|
):
|
||||||
|
m_fetch_ical.side_effect = OSError("iCal unreachable")
|
||||||
|
m_parse_ical.side_effect = OSError("iCal parse error")
|
||||||
|
client = MagicMock()
|
||||||
|
client.get_lessons.side_effect = OSError("Pronote API error")
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
|
||||||
|
with pytest.raises(PipelineCriticalError) as exc_info:
|
||||||
|
mock_fetcher.fetch_agenda()
|
||||||
|
|
||||||
|
assert "les sources ical et pronotepy ont échoué" in str(exc_info.value)
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_agenda_fallback_returns_empty_logs_warning(
|
||||||
|
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
|
||||||
|
) -> None:
|
||||||
|
"""Test que fetch_agenda retourne ([], []) et journalise un avertissement si le repli retourne vide.
|
||||||
|
|
||||||
|
On mock ICAL pour échouer, pronotepy configuré et retourne vide. On vérifie le retour et le log.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:param caplog: Fixture pytest pour capturer les logs.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_fetcher._settings.pronote.agenda_source = "ical"
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
|
patch(
|
||||||
|
"pronote_sync.sources.pronote.fallback.PronoteFetcher._fetch_agenda_pronotepy"
|
||||||
|
) as m_fetch_pronotepy,
|
||||||
|
):
|
||||||
|
m_fetch_ical.side_effect = OSError("iCal unreachable")
|
||||||
|
m_parse_ical.side_effect = OSError("iCal parse error")
|
||||||
|
client = MagicMock()
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
m_fetch_pronotepy.return_value = ([], []) # Empty result from fallback
|
||||||
|
|
||||||
|
lessons, events = mock_fetcher.fetch_agenda()
|
||||||
|
|
||||||
|
assert lessons == []
|
||||||
|
assert events == []
|
||||||
|
assert "a retourné un résultat vide après l'échec de ical" in caplog.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_homework_sources_explicit_ical_mode(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test _homework_sources en mode ICAL.
|
||||||
|
|
||||||
|
On vérifie que la source primaire est ical et que le repli est pronotepy si configuré,
|
||||||
|
ou None sinon.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_fetcher._settings.pronote.homework_source = "ical"
|
||||||
|
|
||||||
|
# With pronotepy configured
|
||||||
|
primary, fallback = mock_fetcher._homework_sources()
|
||||||
|
assert primary == "ical"
|
||||||
|
assert fallback == "pronotepy"
|
||||||
|
|
||||||
|
# Without pronotepy configured
|
||||||
|
mock_fetcher._settings.pronote.pronote_url = None
|
||||||
|
primary, fallback = mock_fetcher._homework_sources()
|
||||||
|
assert primary == "ical"
|
||||||
|
assert fallback is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_homework_sources_explicit_pronotepy_mode(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test _homework_sources en mode PRONOTEPY.
|
||||||
|
|
||||||
|
On vérifie que la source primaire est pronotepy et que le repli est ical si configuré,
|
||||||
|
ou None sinon.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_fetcher._settings.pronote.homework_source = "pronotepy"
|
||||||
|
|
||||||
|
# With ical configured
|
||||||
|
primary, fallback = mock_fetcher._homework_sources()
|
||||||
|
assert primary == "pronotepy"
|
||||||
|
assert fallback == "ical"
|
||||||
|
|
||||||
|
# Without ical configured
|
||||||
|
mock_fetcher._settings.pronote.ical_url = None
|
||||||
|
primary, fallback = mock_fetcher._homework_sources()
|
||||||
|
assert primary == "pronotepy"
|
||||||
|
assert fallback is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_homework_sources_auto_only_pronotepy_configured(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test _homework_sources en mode AUTO avec seulement pronotepy configuré.
|
||||||
|
|
||||||
|
On désactive ical_url mais on garde pronotepy configuré. On vérifie que la
|
||||||
|
source primaire est pronotepy et qu'il n'y a pas de repli.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_fetcher._settings.pronote.homework_source = "auto"
|
||||||
|
mock_fetcher._settings.pronote.ical_url = None
|
||||||
|
primary, fallback = mock_fetcher._homework_sources()
|
||||||
|
assert primary == "pronotepy"
|
||||||
|
assert fallback is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_homework_sources_auto_no_source_configured_raises(mock_fetcher: PronoteFetcher) -> None:
|
||||||
|
"""Test _homework_sources en mode AUTO avec aucune source configurée.
|
||||||
|
|
||||||
|
On désactive les deux sources. On vérifie que l'erreur PipelineCriticalError est levée.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_fetcher._settings.pronote.homework_source = "auto"
|
||||||
|
mock_fetcher._settings.pronote.ical_url = None
|
||||||
|
mock_fetcher._settings.pronote.pronote_url = None
|
||||||
|
|
||||||
|
with pytest.raises(PipelineCriticalError) as exc_info:
|
||||||
|
mock_fetcher._homework_sources()
|
||||||
|
|
||||||
|
assert "ni la source iCal ni pronotepy n'est configurée" in str(exc_info.value)
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_homework_fallback_both_fail_raises_pipeline_critical_error(
|
||||||
|
mock_fetcher: PronoteFetcher,
|
||||||
|
) -> None:
|
||||||
|
"""Test que fetch_homework lève PipelineCriticalError si ICAL échoue et pronotepy aussi.
|
||||||
|
|
||||||
|
On mock ICAL pour échouer, on garde pronotepy configuré mais on mock son échec.
|
||||||
|
On vérifie que l'erreur PipelineCriticalError est levée.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
target_date = date(2025, 9, 10)
|
||||||
|
mock_fetcher._settings.pronote.homework_source = "ical"
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.collect_homeworks") as m_collect,
|
||||||
|
):
|
||||||
|
m_fetch_ical.side_effect = OSError("iCal unreachable")
|
||||||
|
m_parse_ical.side_effect = OSError("iCal parse error")
|
||||||
|
client = MagicMock()
|
||||||
|
client.get_homeworks.side_effect = OSError("Pronote API error")
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
m_collect.side_effect = OSError("collect error")
|
||||||
|
|
||||||
|
with pytest.raises(PipelineCriticalError) as exc_info:
|
||||||
|
mock_fetcher.fetch_homework(target_date)
|
||||||
|
|
||||||
|
assert "les sources ical et pronotepy ont échoué" in str(exc_info.value)
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_homework_fallback_returns_empty_logs_warning(
|
||||||
|
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
|
||||||
|
) -> None:
|
||||||
|
"""Test que fetch_homework retourne [] et journalise un avertissement si le repli retourne vide.
|
||||||
|
|
||||||
|
On mock ICAL pour échouer, pronotepy configuré et retourne vide. On vérifie le retour et le log.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:param caplog: Fixture pytest pour capturer les logs.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
target_date = date(2025, 9, 10)
|
||||||
|
mock_fetcher._settings.pronote.homework_source = "ical"
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
|
patch("pronote_sync.sources.pronote.fallback.collect_homeworks") as m_collect,
|
||||||
|
):
|
||||||
|
m_fetch_ical.side_effect = OSError("iCal unreachable")
|
||||||
|
m_parse_ical.side_effect = OSError("iCal parse error")
|
||||||
|
client = MagicMock()
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
m_collect.return_value = [] # Empty result
|
||||||
|
|
||||||
|
result = mock_fetcher.fetch_homework(target_date)
|
||||||
|
|
||||||
|
assert result == []
|
||||||
|
assert "a retourné un résultat vide après l'échec de ical" in caplog.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_informations_logs_and_re_raises_secret(
|
||||||
|
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
|
||||||
|
) -> None:
|
||||||
|
"""Test que fetch_informations journalise et relance les exceptions avec secret masqué.
|
||||||
|
|
||||||
|
On mock get_informations pour lever une exception contenant une URL secrète.
|
||||||
|
On vérifie que l'exception est relancée et que le log ne contient pas le secret.
|
||||||
|
|
||||||
|
:param mock_fetcher: Fetcher de test.
|
||||||
|
:param caplog: Fixture pytest pour capturer les logs.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
client = MagicMock()
|
||||||
|
error_msg = "Erreur Pronote : impossible de récupérer les informations https://pronote.example.com/infos?token=SECRET_TOKEN_789"
|
||||||
|
client.get_informations.side_effect = OSError(error_msg)
|
||||||
|
mock_fetcher._pronote_client = client
|
||||||
|
|
||||||
|
with pytest.raises(OSError) as exc_info:
|
||||||
|
mock_fetcher.fetch_informations()
|
||||||
|
|
||||||
|
assert exc_info.value is client.get_informations.side_effect
|
||||||
|
assert "SECRET_TOKEN_789" not in caplog.text
|
||||||
|
assert (
|
||||||
|
"pronote.example.com/infos?token=REDACTED" in caplog.text
|
||||||
|
or "pronote.example.com/infos" in caplog.text
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# Ensure trailing newline
|
||||||
600
tests/unit/test_ical.py
Normal file
600
tests/unit/test_ical.py
Normal file
@@ -0,0 +1,600 @@
|
|||||||
|
"""Tests unitaires pour le module iCal : téléchargement et parsing.
|
||||||
|
|
||||||
|
Ce module teste :
|
||||||
|
- La récupération du flux iCal (file://, HTTP)
|
||||||
|
- Le parsing des événements en modèles Lesson, SchoolEvent
|
||||||
|
- L'extraction et normalisation des devoirs
|
||||||
|
- La collecte et déduplication des devoirs par date cible
|
||||||
|
|
||||||
|
Les tests utilisent des mocks pour éviter tout appel réseau réel.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import tempfile
|
||||||
|
import urllib.parse
|
||||||
|
from datetime import date, datetime
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
import responses
|
||||||
|
|
||||||
|
from pronote_sync.models.agenda import HomeworkBlock, Lesson, LessonStatus
|
||||||
|
from pronote_sync.sources.pronote.ical import (
|
||||||
|
collect_homeworks,
|
||||||
|
fetch_ical,
|
||||||
|
generate_homework_id,
|
||||||
|
get_calendar_name,
|
||||||
|
normalize_homework_text,
|
||||||
|
parse_body,
|
||||||
|
parse_ical,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def valid_ical_content() -> str:
|
||||||
|
"""Contenu iCal valide pour tests de parsing."""
|
||||||
|
return """BEGIN:VCALENDAR
|
||||||
|
VERSION:2.0
|
||||||
|
PRODID:-//Test//Test//FR
|
||||||
|
X-WR-CALNAME:Test Calendar
|
||||||
|
BEGIN:VEVENT
|
||||||
|
UID:test-1@test.net
|
||||||
|
DTSTAMP:20260905T120000Z
|
||||||
|
DTSTART:20260905T080000Z
|
||||||
|
DTEND:20260905T090000Z
|
||||||
|
SUMMARY:Math
|
||||||
|
CATEGORIES:Cours
|
||||||
|
DESCRIPTION:<div>Matière : Math\nProfesseur : M. Dupont\nSalle : 204\n<strong>Contenu pédagogique :</strong>Résoudre des équations.</div>
|
||||||
|
END:VEVENT
|
||||||
|
END:VCALENDAR
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def invalid_ical_content() -> str:
|
||||||
|
"""Contenu iCal invalide (sans BEGIN:VCALENDAR)."""
|
||||||
|
return "INVALID:CONTENT\nThis is not a valid iCal file."
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_ical_file_protocol() -> None:
|
||||||
|
"""fetch_ical("file://tests/fixtures/pronote-4e.ics") retourne un contenu commençant par BEGIN:VCALENDAR.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
|
||||||
|
url = f"file://{fixture_path}"
|
||||||
|
|
||||||
|
content = fetch_ical(url)
|
||||||
|
assert content.lstrip().startswith("BEGIN:VCALENDAR")
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_ical_file_uri_decoding() -> None:
|
||||||
|
"""fetch_ical("file://path%20with%20spaces") décode correctement le chemin.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
# Créer un fichier temporaire avec un espace dans le nom
|
||||||
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||||||
|
temp_path = Path(tmpdir) / "fichier avec espaces.ics"
|
||||||
|
temp_path.write_text(
|
||||||
|
"BEGIN:VCALENDAR\nVERSION:2.0\nPRODID:-//Test//Test//FR\nEND:VCALENDAR",
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
|
||||||
|
# URL encodée avec espace
|
||||||
|
encoded_name = urllib.parse.quote("fichier avec espaces.ics")
|
||||||
|
url = f"file://{tmpdir}/{encoded_name}"
|
||||||
|
|
||||||
|
# Cela devrait fonctionner car Path.read_text décode l'URL
|
||||||
|
content = fetch_ical(url)
|
||||||
|
assert content.lstrip().startswith("BEGIN:VCALENDAR")
|
||||||
|
|
||||||
|
|
||||||
|
@responses.activate
|
||||||
|
def test_fetch_ical_invalid_content() -> None:
|
||||||
|
"""Si le contenu ne commence pas par BEGIN:VCALENDAR, une exception est levée.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
responses.add(
|
||||||
|
responses.GET,
|
||||||
|
"https://example.com/ical.ics",
|
||||||
|
body="INVALID:CONTENT",
|
||||||
|
status=200,
|
||||||
|
)
|
||||||
|
|
||||||
|
with pytest.raises(ValueError, match="Flux iCal invalide"):
|
||||||
|
fetch_ical("https://example.com/ical.ics")
|
||||||
|
|
||||||
|
|
||||||
|
@responses.activate
|
||||||
|
def test_fetch_ical_http() -> None:
|
||||||
|
"""Mock de requests.get pour retourner un contenu iCal valide.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
valid_content = "BEGIN:VCALENDAR\nVERSION:2.0\nPRODID:-//Test//Test//FR\nEND:VCALENDAR"
|
||||||
|
|
||||||
|
responses.add(
|
||||||
|
responses.GET,
|
||||||
|
"https://pronote.example.com/ical.ics",
|
||||||
|
body=valid_content,
|
||||||
|
status=200,
|
||||||
|
)
|
||||||
|
|
||||||
|
content = fetch_ical("https://pronote.example.com/ical.ics")
|
||||||
|
assert content == valid_content
|
||||||
|
|
||||||
|
|
||||||
|
@responses.activate
|
||||||
|
def test_fetch_ical_redacts_errors() -> None:
|
||||||
|
"""Les messages d'erreur ne contiennent pas l'URL complète (doit être masquée).
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
responses.add(
|
||||||
|
responses.GET,
|
||||||
|
"https://pronote.example.com/ical.ics",
|
||||||
|
body=Exception("Erreur réseau"),
|
||||||
|
status=500,
|
||||||
|
)
|
||||||
|
|
||||||
|
with pytest.raises(Exception) as exc_info:
|
||||||
|
fetch_ical("https://pronote.example.com/ical.ics?token=secret123")
|
||||||
|
|
||||||
|
error_msg = str(exc_info.value)
|
||||||
|
# Vérifie que le token secret n'est pas dans le message
|
||||||
|
assert "secret123" not in error_msg
|
||||||
|
# Vérifie que l'URL est masquée (utilise redact_url)
|
||||||
|
assert "https://pronote.example.com/ical.ics" in error_msg
|
||||||
|
# Le message doit contenir la partie masquée
|
||||||
|
assert "...ics" in error_msg or "pronote.example.com/ical" in error_msg
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_calendar_name() -> None:
|
||||||
|
"""get_calendar_name(raw_ical) retourne le nom du calendrier.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
|
||||||
|
with open(fixture_path, encoding="utf-8") as f:
|
||||||
|
content = f.read()
|
||||||
|
|
||||||
|
name = get_calendar_name(content)
|
||||||
|
assert name == "Classe de 4e"
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_calendar_name_with_params() -> None:
|
||||||
|
"""Test avec X-WR-CALNAME;LANGUAGE=fr:TestName.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
raw_ical = """BEGIN:VCALENDAR
|
||||||
|
X-WR-CALNAME;LANGUAGE=fr:TestName
|
||||||
|
END:VCALENDAR
|
||||||
|
"""
|
||||||
|
name = get_calendar_name(raw_ical)
|
||||||
|
assert name == "TestName"
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_calendar_name_none() -> None:
|
||||||
|
"""Retourne None si X-WR-CALNAME est absent.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
raw_ical = """BEGIN:VCALENDAR
|
||||||
|
VERSION:2.0
|
||||||
|
END:VCALENDAR
|
||||||
|
"""
|
||||||
|
name = get_calendar_name(raw_ical)
|
||||||
|
assert name is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_ical_returns_lessons() -> None:
|
||||||
|
"""parse_ical(fixture_content) retourne au moins 2 cours.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
|
||||||
|
with open(fixture_path, encoding="utf-8") as f:
|
||||||
|
content = f.read()
|
||||||
|
|
||||||
|
lessons, homeworks, school_events = parse_ical(content)
|
||||||
|
assert len(lessons) >= 2
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_ical_detects_cancelled_course() -> None:
|
||||||
|
"""Un cours a status == LessonStatus.CANCELLED.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
|
||||||
|
with open(fixture_path, encoding="utf-8") as f:
|
||||||
|
content = f.read()
|
||||||
|
|
||||||
|
lessons, _, _ = parse_ical(content)
|
||||||
|
cancelled_lessons = [lesson for lesson in lessons if lesson.status == LessonStatus.CANCELLED]
|
||||||
|
assert len(cancelled_lessons) >= 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_ical_returns_school_events() -> None:
|
||||||
|
"""Au moins 1 événement scolaire est retourné.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
|
||||||
|
with open(fixture_path, encoding="utf-8") as f:
|
||||||
|
content = f.read()
|
||||||
|
|
||||||
|
_, _, school_events = parse_ical(content)
|
||||||
|
assert len(school_events) >= 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_ical_homeworks_empty() -> None:
|
||||||
|
"""La liste des devoirs est toujours vide depuis parse_ical.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
|
||||||
|
with open(fixture_path, encoding="utf-8") as f:
|
||||||
|
content = f.read()
|
||||||
|
|
||||||
|
_, homeworks, _ = parse_ical(content)
|
||||||
|
assert homeworks == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_ical_lesson_fields() -> None:
|
||||||
|
"""Vérifie qu'un cours a les bons champs (matière, profs, salles).
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
|
||||||
|
with open(fixture_path, encoding="utf-8") as f:
|
||||||
|
content = f.read()
|
||||||
|
|
||||||
|
lessons, _, _ = parse_ical(content)
|
||||||
|
assert len(lessons) > 0
|
||||||
|
|
||||||
|
# Vérifions le premier cours (Mathématiques)
|
||||||
|
lesson = lessons[0]
|
||||||
|
assert lesson.subject == "Mathématiques"
|
||||||
|
assert lesson.teachers == ("M. Dupont",)
|
||||||
|
assert lesson.rooms == ("204",)
|
||||||
|
assert lesson.status == LessonStatus.NORMAL
|
||||||
|
|
||||||
|
|
||||||
|
def test_collect_homeworks_dedup() -> None:
|
||||||
|
"""Étant donné des cours avec des blocs de devoirs, collect_homeworks retourne des devoirs dédupliqués.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
# Créer des cours avec des blocs de devoirs en double
|
||||||
|
lesson1 = Lesson(
|
||||||
|
id="lesson1",
|
||||||
|
start=datetime(2026, 9, 10, 8, 0),
|
||||||
|
end=datetime(2026, 9, 10, 9, 0),
|
||||||
|
subject="Math",
|
||||||
|
teachers=("M. Dupont",),
|
||||||
|
rooms=("204",),
|
||||||
|
group=None,
|
||||||
|
status=LessonStatus.NORMAL,
|
||||||
|
content=None,
|
||||||
|
homework_blocks=(
|
||||||
|
HomeworkBlock(
|
||||||
|
kind="due",
|
||||||
|
date=date(2026, 9, 10),
|
||||||
|
text="Exercice 1 à 5 page 42",
|
||||||
|
html="<p>Exercice 1 à 5 page 42</p>",
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
lesson2 = Lesson(
|
||||||
|
id="lesson2",
|
||||||
|
start=datetime(2026, 9, 10, 10, 0),
|
||||||
|
end=datetime(2026, 9, 10, 11, 0),
|
||||||
|
subject="Physique",
|
||||||
|
teachers=("M. Martin",),
|
||||||
|
rooms=("205",),
|
||||||
|
group=None,
|
||||||
|
status=LessonStatus.NORMAL,
|
||||||
|
content=None,
|
||||||
|
homework_blocks=(
|
||||||
|
HomeworkBlock(
|
||||||
|
kind="due",
|
||||||
|
date=date(2026, 9, 10),
|
||||||
|
text="Exercice 1 à 5 page 42",
|
||||||
|
html="<p>Exercice 1 à 5 page 42</p>",
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
homeworks = collect_homeworks([lesson1, lesson2], target_date=date(2026, 9, 10))
|
||||||
|
assert len(homeworks) == 1 # Un seul devoir dédupliqué
|
||||||
|
|
||||||
|
|
||||||
|
def test_collect_homeworks_id_stability() -> None:
|
||||||
|
"""Deux devoirs avec le même texte et date d'échéance produisent le même ID.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
lesson1 = Lesson(
|
||||||
|
id="lesson1",
|
||||||
|
start=datetime(2026, 9, 10, 8, 0),
|
||||||
|
end=datetime(2026, 9, 10, 9, 0),
|
||||||
|
subject="Math",
|
||||||
|
teachers=("M. Dupont",),
|
||||||
|
rooms=("204",),
|
||||||
|
group=None,
|
||||||
|
status=LessonStatus.NORMAL,
|
||||||
|
content=None,
|
||||||
|
homework_blocks=(
|
||||||
|
HomeworkBlock(
|
||||||
|
kind="due",
|
||||||
|
date=date(2026, 9, 10),
|
||||||
|
text="Devoir commun",
|
||||||
|
html="<p>Devoir commun</p>",
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
lesson2 = Lesson(
|
||||||
|
id="lesson2",
|
||||||
|
start=datetime(2026, 9, 10, 10, 0),
|
||||||
|
end=datetime(2026, 9, 10, 11, 0),
|
||||||
|
subject="Physique",
|
||||||
|
teachers=("M. Martin",),
|
||||||
|
rooms=("205",),
|
||||||
|
group=None,
|
||||||
|
status=LessonStatus.NORMAL,
|
||||||
|
content=None,
|
||||||
|
homework_blocks=(
|
||||||
|
HomeworkBlock(
|
||||||
|
kind="due",
|
||||||
|
date=date(2026, 9, 10),
|
||||||
|
text="Devoir commun",
|
||||||
|
html="<p>Devoir commun</p>",
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
homeworks = collect_homeworks([lesson1, lesson2], target_date=date(2026, 9, 10))
|
||||||
|
assert len(homeworks) == 1
|
||||||
|
assert homeworks[0].id == generate_homework_id(date(2026, 9, 10), "devoir commun")
|
||||||
|
|
||||||
|
|
||||||
|
def test_collect_homeworks_sorted() -> None:
|
||||||
|
"""Les résultats sont triés par (subject.lower(), text.lower()).
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
lesson1 = Lesson(
|
||||||
|
id="lesson1",
|
||||||
|
start=datetime(2026, 9, 10, 8, 0),
|
||||||
|
end=datetime(2026, 9, 10, 9, 0),
|
||||||
|
subject="Zoologie",
|
||||||
|
teachers=("M. A",),
|
||||||
|
rooms=("204",),
|
||||||
|
group=None,
|
||||||
|
status=LessonStatus.NORMAL,
|
||||||
|
content=None,
|
||||||
|
homework_blocks=(
|
||||||
|
HomeworkBlock(
|
||||||
|
kind="due", date=date(2026, 9, 10), text="Devoir B", html="<p>Devoir B</p>"
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
lesson2 = Lesson(
|
||||||
|
id="lesson2",
|
||||||
|
start=datetime(2026, 9, 10, 10, 0),
|
||||||
|
end=datetime(2026, 9, 10, 11, 0),
|
||||||
|
subject="Mathématiques",
|
||||||
|
teachers=("M. B",),
|
||||||
|
rooms=("205",),
|
||||||
|
group=None,
|
||||||
|
status=LessonStatus.NORMAL,
|
||||||
|
content=None,
|
||||||
|
homework_blocks=(
|
||||||
|
HomeworkBlock(
|
||||||
|
kind="due", date=date(2026, 9, 10), text="Devoir A", html="<p>Devoir A</p>"
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
homeworks = collect_homeworks([lesson1, lesson2], target_date=date(2026, 9, 10))
|
||||||
|
assert len(homeworks) == 2
|
||||||
|
assert homeworks[0].subject == "Mathématiques"
|
||||||
|
assert homeworks[0].text == "Devoir A"
|
||||||
|
assert homeworks[1].subject == "Zoologie"
|
||||||
|
assert homeworks[1].text == "Devoir B"
|
||||||
|
|
||||||
|
|
||||||
|
def test_collect_homeworks_empty() -> None:
|
||||||
|
"""Aucun bloc de devoir → liste vide.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
lesson = Lesson(
|
||||||
|
id="lesson1",
|
||||||
|
start=datetime(2026, 9, 10, 8, 0),
|
||||||
|
end=datetime(2026, 9, 10, 9, 0),
|
||||||
|
subject="Math",
|
||||||
|
teachers=("M. Dupont",),
|
||||||
|
rooms=("204",),
|
||||||
|
group=None,
|
||||||
|
status=LessonStatus.NORMAL,
|
||||||
|
content=None,
|
||||||
|
)
|
||||||
|
|
||||||
|
homeworks = collect_homeworks([lesson], target_date=date(2026, 9, 10))
|
||||||
|
assert homeworks == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_normalize_homework_text() -> None:
|
||||||
|
"""Vérifie la normalisation des espaces, suppression HTML et minuscules.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
text = " <p>Exercice 1 à 5</p> \n\n page 42 "
|
||||||
|
normalized = normalize_homework_text(text)
|
||||||
|
assert normalized == "exercice 1 à 5 page 42"
|
||||||
|
|
||||||
|
|
||||||
|
def test_generate_homework_id_format() -> None:
|
||||||
|
"""Retourne un ID de 12 caractères hexadécimaux.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
due_on = date(2026, 9, 10)
|
||||||
|
text = "devoir test"
|
||||||
|
homework_id = generate_homework_id(due_on, text)
|
||||||
|
assert len(homework_id) == 12
|
||||||
|
assert all(c in "0123456789abcdef" for c in homework_id)
|
||||||
|
|
||||||
|
|
||||||
|
def test_generate_homework_id_deterministic() -> None:
|
||||||
|
"""Mêmes entrées → même sortie.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
due_on = date(2026, 9, 10)
|
||||||
|
text = "devoir commun"
|
||||||
|
id1 = generate_homework_id(due_on, text)
|
||||||
|
id2 = generate_homework_id(due_on, text)
|
||||||
|
assert id1 == id2
|
||||||
|
|
||||||
|
|
||||||
|
CANCELLED_STATUS_ICAL = """BEGIN:VCALENDAR
|
||||||
|
VERSION:2.0
|
||||||
|
X-WR-CALNAME:Test
|
||||||
|
BEGIN:VEVENT
|
||||||
|
UID:Test-123-20260906T120000Z-Index-Education
|
||||||
|
DTSTART:20260907T080000Z
|
||||||
|
DTEND:20260907T090000Z
|
||||||
|
SUMMARY:Test Course
|
||||||
|
STATUS:CANCELLED
|
||||||
|
DESCRIPTION:<div></div>
|
||||||
|
END:VEVENT
|
||||||
|
END:VCALENDAR"""
|
||||||
|
|
||||||
|
|
||||||
|
NORMAL_STATUS_ICAL = """BEGIN:VCALENDAR
|
||||||
|
VERSION:2.0
|
||||||
|
X-WR-CALNAME:Test
|
||||||
|
BEGIN:VEVENT
|
||||||
|
UID:Test-123-20260906T120000Z-Index-Education
|
||||||
|
DTSTART:20260907T080000Z
|
||||||
|
DTEND:20260907T090000Z
|
||||||
|
SUMMARY:Test Course
|
||||||
|
STATUS:CONFIRMED
|
||||||
|
DESCRIPTION:<div></div>
|
||||||
|
END:VEVENT
|
||||||
|
END:VCALENDAR"""
|
||||||
|
|
||||||
|
|
||||||
|
MOVED_BY_CATEGORY_ICAL = """BEGIN:VCALENDAR
|
||||||
|
VERSION:2.0
|
||||||
|
X-WR-CALNAME:Test
|
||||||
|
BEGIN:VEVENT
|
||||||
|
UID:Test-123-20260906T120000Z-Index-Education
|
||||||
|
DTSTART:20260907T080000Z
|
||||||
|
DTEND:20260907T090000Z
|
||||||
|
SUMMARY:Test Course
|
||||||
|
CATEGORIES:Cours - Cours déplacé
|
||||||
|
DESCRIPTION:<div></div>
|
||||||
|
END:VEVENT
|
||||||
|
END:VCALENDAR"""
|
||||||
|
|
||||||
|
|
||||||
|
MULTIPLE_BLOCKS_SAME_DATE_ICAL = """BEGIN:VCALENDAR
|
||||||
|
VERSION:2.0
|
||||||
|
X-WR-CALNAME:Test
|
||||||
|
BEGIN:VEVENT
|
||||||
|
UID:Test-123-20260906T120000Z-Index-Education
|
||||||
|
DTSTART:20260907T080000Z
|
||||||
|
DTEND:20260907T090000Z
|
||||||
|
SUMMARY:Test Course
|
||||||
|
CATEGORIES:Cours
|
||||||
|
DESCRIPTION:<div>
|
||||||
|
Matière : Math
|
||||||
|
Professeur : M. Dupont
|
||||||
|
Salle : 204
|
||||||
|
|
||||||
|
<strong>Pour le 10/09/2026 :</strong>
|
||||||
|
Exercice 1 à 5 page 42.
|
||||||
|
<strong>Pour le 10/09/2026 :</strong>
|
||||||
|
Exercice 6 à 10 page 43.
|
||||||
|
</div>
|
||||||
|
END:VEVENT
|
||||||
|
END:VCALENDAR"""
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_ical_status_cancelled_only() -> None:
|
||||||
|
"""Un cours avec STATUS:CANCELLED mais sans CATEGORIES contenant 'Cours annulé' a status == LessonStatus.CANCELLED.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
lessons, _, _ = parse_ical(CANCELLED_STATUS_ICAL)
|
||||||
|
assert len(lessons) == 1
|
||||||
|
assert lessons[0].status == LessonStatus.CANCELLED
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_ical_status_normal_without_cancel() -> None:
|
||||||
|
"""Un cours avec STATUS:CONFIRMED (ou sans STATUS) a status == LessonStatus.NORMAL.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
lessons, _, _ = parse_ical(NORMAL_STATUS_ICAL)
|
||||||
|
assert len(lessons) == 1
|
||||||
|
assert lessons[0].status == LessonStatus.NORMAL
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_ical_moved_by_category_only() -> None:
|
||||||
|
"""Un cours avec CATEGORIES:Cours - Cours déplacé et sans STATUS a status == LessonStatus.MOVED.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
lessons, _, _ = parse_ical(MOVED_BY_CATEGORY_ICAL)
|
||||||
|
assert len(lessons) == 1
|
||||||
|
assert lessons[0].status == LessonStatus.MOVED
|
||||||
|
|
||||||
|
|
||||||
|
def test_parse_body_multiple_blocks_same_date() -> None:
|
||||||
|
"""Un DESCRIPTION avec deux sections 'Pour le' à la même date conserve les deux blocs.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
body_html = (
|
||||||
|
"<div>\n"
|
||||||
|
" <strong>Pour le 10/09/2026:</strong>\n"
|
||||||
|
" Exercice 1 à 5 page 42.\n"
|
||||||
|
" <strong>Pour le 10/09/2026:</strong>\n"
|
||||||
|
" Exercice 6 à 10 page 43.\n"
|
||||||
|
"</div>"
|
||||||
|
)
|
||||||
|
content, due_blocks, assigned_blocks = parse_body(body_html)
|
||||||
|
assert len(due_blocks) == 2
|
||||||
|
assert due_blocks[0][0] == date(2026, 9, 10)
|
||||||
|
assert due_blocks[0][1] == "Exercice 1 à 5 page 42."
|
||||||
|
assert due_blocks[1][0] == date(2026, 9, 10)
|
||||||
|
assert due_blocks[1][1] == "Exercice 6 à 10 page 43."
|
||||||
|
|
||||||
|
|
||||||
|
def test_collect_homeworks_from_fixture() -> None:
|
||||||
|
"""Parse le fixture pronote-4e.ics, appelle collect_homeworks pour le 10/09/2026 et vérifie qu'au moins un devoir est retourné.
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
|
||||||
|
with open(fixture_path, encoding="utf-8") as f:
|
||||||
|
content = f.read()
|
||||||
|
|
||||||
|
lessons, _, _ = parse_ical(content)
|
||||||
|
homeworks = collect_homeworks(lessons, date(2026, 9, 10))
|
||||||
|
|
||||||
|
assert len(homeworks) >= 1
|
||||||
|
# Vérifie qu'au moins un devoir a le bon sujet et texte
|
||||||
|
assert any(hw.subject == "Mathématiques" for hw in homeworks)
|
||||||
|
assert any("Exercices 1 à 5 page 42" in hw.text for hw in homeworks)
|
||||||
439
tests/unit/test_pronote_client.py
Normal file
439
tests/unit/test_pronote_client.py
Normal file
@@ -0,0 +1,439 @@
|
|||||||
|
"""Tests unitaires pour le client Pronote via pronotepy.
|
||||||
|
|
||||||
|
Ce module vérifie le comportement du client ``PronoteClient`` et de son
|
||||||
|
protocole ``PronoteClientProtocol``. Tous les tests sont unitaires et
|
||||||
|
utilisent des mocks pour éviter tout accès réseau réel à Pronote.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from datetime import date, datetime
|
||||||
|
|
||||||
|
import pronotepy
|
||||||
|
import pytest
|
||||||
|
import pytest_mock
|
||||||
|
from pydantic import SecretStr
|
||||||
|
|
||||||
|
from pronote_sync.config.settings import PronoteSettings
|
||||||
|
from pronote_sync.models.agenda import Lesson, LessonStatus
|
||||||
|
from pronote_sync.models.homework import Homework
|
||||||
|
from pronote_sync.models.message import Message, MessageType
|
||||||
|
from pronote_sync.sources.pronote.client import PronoteClient, PronoteClientProtocol
|
||||||
|
|
||||||
|
# --- Protocol tests ---
|
||||||
|
|
||||||
|
|
||||||
|
def test_protocol_methods(mocker: pytest_mock.MockerFixture) -> None:
|
||||||
|
"""Vérifie que le protocole PronoteClientProtocol expose les méthodes attendues.
|
||||||
|
|
||||||
|
:param mocker: Fixture pytest-mock pour le mocking.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
assert hasattr(PronoteClientProtocol, "get_messages")
|
||||||
|
assert hasattr(PronoteClientProtocol, "get_informations")
|
||||||
|
assert hasattr(PronoteClientProtocol, "get_lessons")
|
||||||
|
assert hasattr(PronoteClientProtocol, "get_homeworks")
|
||||||
|
|
||||||
|
|
||||||
|
# --- Client with mocked pronotepy ---
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def pronote_settings() -> PronoteSettings:
|
||||||
|
"""Fournit des paramètres Pronote valides pour les tests.
|
||||||
|
|
||||||
|
:return: Instance de PronoteSettings avec des valeurs par défaut.
|
||||||
|
:rtype: PronoteSettings
|
||||||
|
"""
|
||||||
|
return PronoteSettings(
|
||||||
|
pronote_url="https://pronote.example.com",
|
||||||
|
username="testuser",
|
||||||
|
password=SecretStr("testpass"),
|
||||||
|
ent="bordeaux",
|
||||||
|
account_type="parent",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def empty_pronote_settings() -> PronoteSettings:
|
||||||
|
"""Fournit des paramètres Pronote vides pour les tests.
|
||||||
|
|
||||||
|
:return: Instance de PronoteSettings avec tous les champs à None.
|
||||||
|
:rtype: PronoteSettings
|
||||||
|
"""
|
||||||
|
return PronoteSettings(username=None, password=None, ent=None)
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_messages_success(
|
||||||
|
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||||
|
) -> None:
|
||||||
|
"""Vérifie que get_messages retourne une liste de Message en cas de succès.
|
||||||
|
|
||||||
|
:param mocker: Fixture pytest-mock pour le mocking.
|
||||||
|
:param pronote_settings: Paramètres Pronote valides.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
# Mock du client pronotepy
|
||||||
|
mock_client = mocker.MagicMock()
|
||||||
|
mock_discussion = mocker.MagicMock()
|
||||||
|
mock_discussion.subject = "Test Subject"
|
||||||
|
mock_message = mocker.MagicMock()
|
||||||
|
mock_message.id = "msg-123"
|
||||||
|
mock_message.content = "Test message content"
|
||||||
|
mock_message.author = "Teacher Test"
|
||||||
|
mock_message.created = datetime(2024, 9, 1, 10, 0, 0)
|
||||||
|
mock_message.seen = True
|
||||||
|
mock_discussion.messages = [mock_message]
|
||||||
|
mock_client.discussions.return_value = [mock_discussion]
|
||||||
|
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||||
|
|
||||||
|
client = PronoteClient(pronote_settings)
|
||||||
|
messages = client.get_messages()
|
||||||
|
|
||||||
|
assert isinstance(messages, list)
|
||||||
|
assert len(messages) == 1
|
||||||
|
message = messages[0]
|
||||||
|
assert isinstance(message, Message)
|
||||||
|
assert message.id == "msg-123"
|
||||||
|
assert message.type == MessageType.DISCUSSION
|
||||||
|
assert message.title == "Test Subject"
|
||||||
|
assert message.content == "Test message content"
|
||||||
|
assert message.author == "Teacher Test"
|
||||||
|
assert message.date == datetime(2024, 9, 1, 10, 0, 0)
|
||||||
|
assert message.read is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_messages_empty_on_error(
|
||||||
|
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||||
|
) -> None:
|
||||||
|
"""Vérifie que get_messages retourne une liste vide en cas d'erreur API.
|
||||||
|
|
||||||
|
:param mocker: Fixture pytest-mock pour le mocking.
|
||||||
|
:param pronote_settings: Paramètres Pronote valides.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_client = mocker.MagicMock()
|
||||||
|
mock_client.discussions.side_effect = pronotepy.PronoteAPIError("API error")
|
||||||
|
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||||
|
|
||||||
|
client = PronoteClient(pronote_settings)
|
||||||
|
messages = client.get_messages()
|
||||||
|
|
||||||
|
assert messages == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_informations_success(
|
||||||
|
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||||
|
) -> None:
|
||||||
|
"""Vérifie que get_informations retourne une liste de Message en cas de succès.
|
||||||
|
|
||||||
|
:param mocker: Fixture pytest-mock pour le mocking.
|
||||||
|
:param pronote_settings: Paramètres Pronote valides.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_client = mocker.MagicMock()
|
||||||
|
mock_info = mocker.MagicMock()
|
||||||
|
mock_info.id = "info-456"
|
||||||
|
mock_info.title = "Important Info"
|
||||||
|
mock_info.content.return_value = "Important content"
|
||||||
|
mock_info.author = "Admin"
|
||||||
|
mock_info.creation_date = datetime(2024, 9, 2, 14, 30, 0)
|
||||||
|
mock_info.read = False
|
||||||
|
mock_info.survey = True
|
||||||
|
mock_client.information_and_surveys.return_value = [mock_info]
|
||||||
|
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||||
|
|
||||||
|
client = PronoteClient(pronote_settings)
|
||||||
|
messages = client.get_informations()
|
||||||
|
|
||||||
|
assert isinstance(messages, list)
|
||||||
|
assert len(messages) == 1
|
||||||
|
message = messages[0]
|
||||||
|
assert isinstance(message, Message)
|
||||||
|
assert message.id == "info-456"
|
||||||
|
assert message.type == MessageType.SURVEY
|
||||||
|
assert message.title == "Important Info"
|
||||||
|
assert message.content == "Important content"
|
||||||
|
assert message.author == "Admin"
|
||||||
|
assert message.date == datetime(2024, 9, 2, 14, 30, 0)
|
||||||
|
assert message.read is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_informations_empty_on_error(
|
||||||
|
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||||
|
) -> None:
|
||||||
|
"""Vérifie que get_informations retourne une liste vide en cas d'erreur API.
|
||||||
|
|
||||||
|
:param mocker: Fixture pytest-mock pour le mocking.
|
||||||
|
:param pronote_settings: Paramètres Pronote valides.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_client = mocker.MagicMock()
|
||||||
|
mock_client.information_and_surveys.side_effect = pronotepy.PronoteAPIError("API error")
|
||||||
|
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||||
|
|
||||||
|
client = PronoteClient(pronote_settings)
|
||||||
|
messages = client.get_informations()
|
||||||
|
|
||||||
|
assert messages == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_lessons_success(
|
||||||
|
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||||
|
) -> None:
|
||||||
|
"""Vérifie que get_lessons retourne une liste de Lesson en cas de succès.
|
||||||
|
|
||||||
|
:param mocker: Fixture pytest-mock pour le mocking.
|
||||||
|
:param pronote_settings: Paramètres Pronote valides.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_client = mocker.MagicMock()
|
||||||
|
|
||||||
|
# Mock des cours
|
||||||
|
mock_lesson = mocker.MagicMock()
|
||||||
|
mock_lesson.id = "lesson-789"
|
||||||
|
mock_lesson.start = datetime(2024, 9, 1, 8, 0, 0)
|
||||||
|
mock_lesson.end = datetime(2024, 9, 1, 9, 30, 0)
|
||||||
|
mock_lesson.subject = mocker.MagicMock()
|
||||||
|
mock_lesson.subject.name = "Maths"
|
||||||
|
mock_lesson.teacher_names = ["Prof A", "Prof B"]
|
||||||
|
mock_lesson.classrooms = ["Salle 101", "Salle 102"]
|
||||||
|
mock_lesson.group_name = "Classe 1"
|
||||||
|
mock_lesson.canceled = False
|
||||||
|
mock_content = mocker.MagicMock()
|
||||||
|
mock_content.description = "Lesson content"
|
||||||
|
mock_lesson.content = mock_content
|
||||||
|
|
||||||
|
mock_client.lessons.return_value = [mock_lesson]
|
||||||
|
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||||
|
|
||||||
|
client = PronoteClient(pronote_settings)
|
||||||
|
lessons = client.get_lessons(date(2024, 9, 1), date(2024, 9, 30))
|
||||||
|
|
||||||
|
assert isinstance(lessons, list)
|
||||||
|
assert len(lessons) == 1
|
||||||
|
lesson = lessons[0]
|
||||||
|
assert isinstance(lesson, Lesson)
|
||||||
|
assert lesson.id == "lesson-789"
|
||||||
|
assert lesson.start == datetime(2024, 9, 1, 8, 0, 0)
|
||||||
|
assert lesson.end == datetime(2024, 9, 1, 9, 30, 0)
|
||||||
|
assert lesson.subject == "Maths"
|
||||||
|
assert lesson.teachers == ("Prof A", "Prof B")
|
||||||
|
assert lesson.rooms == ("Salle 101", "Salle 102")
|
||||||
|
assert lesson.group == "Classe 1"
|
||||||
|
assert lesson.status == LessonStatus.NORMAL
|
||||||
|
assert lesson.content == "Lesson content"
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_homeworks_success(
|
||||||
|
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||||
|
) -> None:
|
||||||
|
"""Vérifie que get_homeworks retourne une liste de Homework en cas de succès.
|
||||||
|
|
||||||
|
:param mocker: Fixture pytest-mock pour le mocking.
|
||||||
|
:param pronote_settings: Paramètres Pronote valides.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_client = mocker.MagicMock()
|
||||||
|
|
||||||
|
# Mock des devoirs
|
||||||
|
mock_hw = mocker.MagicMock()
|
||||||
|
mock_hw.id = "hw-101"
|
||||||
|
mock_hw.subject = mocker.MagicMock()
|
||||||
|
mock_hw.subject.name = "Maths"
|
||||||
|
mock_hw.date = date(2024, 9, 15)
|
||||||
|
mock_hw.description = "Do your homework"
|
||||||
|
|
||||||
|
mock_client.homework.return_value = [mock_hw]
|
||||||
|
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||||
|
|
||||||
|
client = PronoteClient(pronote_settings)
|
||||||
|
homeworks = client.get_homeworks(date(2024, 9, 1), date(2024, 9, 30))
|
||||||
|
|
||||||
|
assert isinstance(homeworks, list)
|
||||||
|
assert len(homeworks) == 1
|
||||||
|
homework = homeworks[0]
|
||||||
|
assert isinstance(homework, Homework)
|
||||||
|
assert homework.id == "hw-101"
|
||||||
|
assert homework.subject == "Maths"
|
||||||
|
assert homework.teachers == ()
|
||||||
|
assert homework.assigned_on is None
|
||||||
|
assert homework.due_on == date(2024, 9, 15)
|
||||||
|
assert homework.text == "Do your homework"
|
||||||
|
assert homework.html == "Do your homework"
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_lessons_propagates_error(
|
||||||
|
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||||
|
) -> None:
|
||||||
|
"""Vérifie que get_lessons propage les exceptions API.
|
||||||
|
|
||||||
|
:param mocker: Fixture pytest-mock pour le mocking.
|
||||||
|
:param pronote_settings: Paramètres Pronote valides.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_client = mocker.MagicMock()
|
||||||
|
mock_client.lessons.side_effect = pronotepy.PronoteAPIError("API error")
|
||||||
|
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||||
|
|
||||||
|
client = PronoteClient(pronote_settings)
|
||||||
|
|
||||||
|
with pytest.raises(pronotepy.PronoteAPIError):
|
||||||
|
client.get_lessons(date(2024, 9, 1), date(2024, 9, 30))
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_homeworks_propagates_error(
|
||||||
|
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||||
|
) -> None:
|
||||||
|
"""Vérifie que get_homeworks propage les exceptions API.
|
||||||
|
|
||||||
|
:param mocker: Fixture pytest-mock pour le mocking.
|
||||||
|
:param pronote_settings: Paramètres Pronote valides.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_client = mocker.MagicMock()
|
||||||
|
mock_client.homework.side_effect = pronotepy.PronoteAPIError("API error")
|
||||||
|
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||||
|
|
||||||
|
client = PronoteClient(pronote_settings)
|
||||||
|
|
||||||
|
with pytest.raises(pronotepy.PronoteAPIError):
|
||||||
|
client.get_homeworks(date(2024, 9, 1), date(2024, 9, 30))
|
||||||
|
|
||||||
|
|
||||||
|
def test_missing_credentials_raises(empty_pronote_settings: PronoteSettings) -> None:
|
||||||
|
"""Vérifie que les appels échouent avec ValueError si les identifiants sont manquants.
|
||||||
|
|
||||||
|
:param empty_pronote_settings: Paramètres Pronote avec tous les champs à None.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
client = PronoteClient(empty_pronote_settings)
|
||||||
|
|
||||||
|
with pytest.raises(ValueError, match="pronote_url, username, password et ent sont requis"):
|
||||||
|
client._connect()
|
||||||
|
|
||||||
|
|
||||||
|
def test_connect_with_ent_resolution(mocker: pytest_mock.MockerFixture) -> None:
|
||||||
|
"""Vérifie que _resolve_ent retourne le callable attendu pour un ENT connu.
|
||||||
|
|
||||||
|
:param mocker: Fixture pytest-mock pour le mocking.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
from pronote_sync.sources.pronote.client import _resolve_ent
|
||||||
|
|
||||||
|
resolver = _resolve_ent("bordeaux")
|
||||||
|
assert resolver is not None
|
||||||
|
|
||||||
|
|
||||||
|
def test_connect_with_unknown_ent_raises(mocker: pytest_mock.MockerFixture) -> None:
|
||||||
|
"""Vérifie que _resolve_ent lève ValueError pour un ENT inconnu.
|
||||||
|
|
||||||
|
:param mocker: Fixture pytest-mock pour le mocking.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
from pronote_sync.sources.pronote.client import _resolve_ent
|
||||||
|
|
||||||
|
with pytest.raises(ValueError) as exc_info:
|
||||||
|
_resolve_ent("inconnu")
|
||||||
|
assert "ENT inconnu : 'inconnu'" in str(exc_info.value)
|
||||||
|
assert "ENT supportés :" in str(exc_info.value)
|
||||||
|
|
||||||
|
|
||||||
|
def test_connect_parent_account_type(
|
||||||
|
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||||
|
) -> None:
|
||||||
|
"""Vérifie que account_type='parent' utilise pronotepy.ParentClient.
|
||||||
|
|
||||||
|
:param mocker: Fixture pytest-mock pour le mocking.
|
||||||
|
:param pronote_settings: Paramètres Pronote valides.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
from unittest.mock import Mock
|
||||||
|
|
||||||
|
from pronote_sync.sources.pronote.client import PronoteClient
|
||||||
|
|
||||||
|
mock_client = mocker.MagicMock()
|
||||||
|
mock_client_class = Mock(return_value=mock_client)
|
||||||
|
mocker.patch("pronotepy.ParentClient", new=mock_client_class)
|
||||||
|
mocker.patch("pronotepy.Client")
|
||||||
|
|
||||||
|
client = PronoteClient(pronote_settings)
|
||||||
|
_ = client._connect()
|
||||||
|
|
||||||
|
# Verify ParentClient was used
|
||||||
|
assert mock_client_class.call_count == 1
|
||||||
|
pronotepy.Client.assert_not_called() # type: ignore[attr-defined]
|
||||||
|
|
||||||
|
|
||||||
|
def test_connect_student_account_type(
|
||||||
|
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||||
|
) -> None:
|
||||||
|
"""Vérifie que account_type='student' utilise pronotepy.Client.
|
||||||
|
|
||||||
|
:param mocker: Fixture pytest-mock pour le mocking.
|
||||||
|
:param pronote_settings: Paramètres Pronote valides.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
from unittest.mock import Mock
|
||||||
|
|
||||||
|
from pronote_sync.sources.pronote.client import PronoteClient
|
||||||
|
|
||||||
|
pronote_settings_student = PronoteSettings(
|
||||||
|
pronote_url="https://pronote.example.com",
|
||||||
|
username="testuser",
|
||||||
|
password=SecretStr("testpass"),
|
||||||
|
ent="bordeaux",
|
||||||
|
account_type="student",
|
||||||
|
)
|
||||||
|
mock_client = mocker.MagicMock()
|
||||||
|
mock_client_class = Mock(return_value=mock_client)
|
||||||
|
mocker.patch("pronotepy.Client", new=mock_client_class)
|
||||||
|
mocker.patch("pronotepy.ParentClient")
|
||||||
|
|
||||||
|
client = PronoteClient(pronote_settings_student)
|
||||||
|
_ = client._connect()
|
||||||
|
|
||||||
|
# Verify Client was used
|
||||||
|
assert mock_client_class.call_count == 1
|
||||||
|
pronotepy.ParentClient.assert_not_called() # type: ignore[attr-defined]
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_messages_degraded_on_error(
|
||||||
|
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||||
|
) -> None:
|
||||||
|
"""Vérifie que get_messages retourne une liste vide en cas d'erreur réseau.
|
||||||
|
|
||||||
|
:param mocker: Fixture pytest-mock pour le mocking.
|
||||||
|
:param pronote_settings: Paramètres Pronote valides.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_client = mocker.MagicMock()
|
||||||
|
mock_client.discussions.side_effect = ConnectionError("Network error")
|
||||||
|
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||||
|
|
||||||
|
client = PronoteClient(pronote_settings)
|
||||||
|
messages = client.get_messages()
|
||||||
|
|
||||||
|
assert messages == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_informations_degraded_on_error(
|
||||||
|
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||||
|
) -> None:
|
||||||
|
"""Vérifie que get_informations retourne une liste vide en cas d'erreur réseau.
|
||||||
|
|
||||||
|
:param mocker: Fixture pytest-mock pour le mocking.
|
||||||
|
:param pronote_settings: Paramètres Pronote valides.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
mock_client = mocker.MagicMock()
|
||||||
|
mock_client.information_and_surveys.side_effect = TimeoutError("Timeout")
|
||||||
|
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||||
|
|
||||||
|
client = PronoteClient(pronote_settings)
|
||||||
|
messages = client.get_informations()
|
||||||
|
|
||||||
|
assert messages == []
|
||||||
|
|
||||||
|
|
||||||
|
# Ensure trailing newline
|
||||||
Reference in New Issue
Block a user