diff --git a/.env.example b/.env.example index 981af19..12a63c5 100644 --- a/.env.example +++ b/.env.example @@ -1,5 +1,7 @@ # --- Pronote --- PRONOTE_ICAL_URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024 +PRONOTE_URL=https://college.ent/pronote/eleve.html +PRONOTE_ACCOUNT_TYPE=parent PRONOTE_USERNAME=parent.dupont PRONOTE_PASSWORD=your_secure_password PRONOTE_ENT=monbureaunumerique diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 4f34eaa..eaebacb 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -26,7 +26,7 @@ repos: name: mypy entry: mypy language: python - additional_dependencies: ["mypy>=1.10.0", "pydantic>=2.0.0", "pydantic-settings>=2.0.0", "pytest>=8.0.0"] + additional_dependencies: ["mypy>=1.10.0", "pydantic>=2.0.0", "pydantic-settings>=2.0.0", "pytest>=8.0.0", "types-requests>=2.31.0", "icalendar>=5.0.0", "pronotepy>=2.15.0", "responses>=0.25.0", "pytest-mock>=3.10.0"] types: [python] pass_filenames: true diff --git a/.secrets.baseline b/.secrets.baseline index fbd8007..4a25695 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -139,11 +139,11 @@ "type": "Hex High Entropy String", "filename": "GUIDE_DEV_PYTHON.md", "hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa", - "is_secret": false, "is_verified": true, - "line_number": 5117 + "line_number": 4893, + "is_secret": false } ] }, - "generated_at": "2026-09-05T21:51:55Z" + "generated_at": "2026-09-06T14:37:44Z" } diff --git a/pronote_sync/config/settings.py b/pronote_sync/config/settings.py index f4e865f..898885d 100644 --- a/pronote_sync/config/settings.py +++ b/pronote_sync/config/settings.py @@ -27,6 +27,8 @@ class PronoteSettings(BaseSettings): username: str | None = None password: SecretStr | None = None ent: str | None = None + pronote_url: str | None = None + account_type: Literal["student", "parent"] = "parent" agenda_source: Literal["auto", "ical", "pronotepy"] = "auto" homework_source: Literal["auto", "ical", "pronotepy"] = "auto" messages_source: Literal["pronotepy"] = "pronotepy" diff --git a/pronote_sync/errors.py b/pronote_sync/errors.py new file mode 100644 index 0000000..e1f17f0 --- /dev/null +++ b/pronote_sync/errors.py @@ -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) diff --git a/pronote_sync/sources/pronote/client.py b/pronote_sync/sources/pronote/client.py new file mode 100644 index 0000000..547d4e9 --- /dev/null +++ b/pronote_sync/sources/pronote/client.py @@ -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 diff --git a/pronote_sync/sources/pronote/fallback.py b/pronote_sync/sources/pronote/fallback.py new file mode 100644 index 0000000..693288b --- /dev/null +++ b/pronote_sync/sources/pronote/fallback.py @@ -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 diff --git a/pronote_sync/sources/pronote/ical.py b/pronote_sync/sources/pronote/ical.py new file mode 100644 index 0000000..3a7e7d5 --- /dev/null +++ b/pronote_sync/sources/pronote/ical.py @@ -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"Contenu pédagogique\s*:\s*(.*?)(?=|\s*$|\Z)", + re.DOTALL, +) +_DUE_PATTERN = re.compile( + r"Pour le (\d{2}/\d{2}/\d{4})\s*:\s*(.*?)(?=|\s*$|\Z)", + re.DOTALL, +) +_ASSIGNED_PATTERN = re.compile( + r"Donné le (\d{2}/\d{2}/\d{4})\s*:\s*(.*?)(?=|\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 ````, le corps + HTML commence à partir du premier ````. + + :param description: Contenu brut de la DESCRIPTION d'un VEVENT. + :return: Tuple ``(en-tête, corps)`` ; le corps est vide si aucun ````. + :rtype: tuple[str, str] + """ + strong_start = description.find("") + 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 ````). + :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 ``Contenu pédagogique :``. + Les devoirs à faire sont extraits des sections ``Pour le JJ/MM/AAAA :`` + (liste de tuples ``(date, texte)`` dans l'ordre du flux) et les devoirs donnés + des sections ``Donné le JJ/MM/AAAA :`` (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 ````). + :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 diff --git a/tests/fixtures/pronote-4e.ics b/tests/fixtures/pronote-4e.ics new file mode 100644 index 0000000..80a48ff --- /dev/null +++ b/tests/fixtures/pronote-4e.ics @@ -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:
+ Matière : Mathématiques + Professeur : M. Dupont + Salle : 204 + Groupe : Classe entière + + Contenu pédagogique : + + Résoudre des équations du second degré. + Pour le 10/09/2026 : + + Exercices 1 à 5 page 42. + Donné le 05/09/2026 : + + Exercices 1 à 5 page 42. +
+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:
+ Matière : Français + Professeur : Mme Martin + Salle : 205 + Groupe : Classe entière + + Contenu pédagogique : + + Étude d'un texte littéraire. +
+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 diff --git a/tests/unit/test_fallback.py b/tests/unit/test_fallback.py new file mode 100644 index 0000000..57c502b --- /dev/null +++ b/tests/unit/test_fallback.py @@ -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="

Devoir sur les fonctions

", + ), + ), + ) + ] + + 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="

Devoir sur les fonctions

", + ) + ] + + 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="

Devoir pour aujourd'hui

", + ), + Homework( + id="hw2", + subject="Physique", + teachers=(), + assigned_on=None, + due_on=other_date, + text="Devoir pour demain", + html="

Devoir pour demain

", + ), + ] + + 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 diff --git a/tests/unit/test_ical.py b/tests/unit/test_ical.py new file mode 100644 index 0000000..0613ed5 --- /dev/null +++ b/tests/unit/test_ical.py @@ -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:
Matière : Math\nProfesseur : M. Dupont\nSalle : 204\nContenu pédagogique :Résoudre des équations.
+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="

Exercice 1 à 5 page 42

", + ), + ), + ) + + 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="

Exercice 1 à 5 page 42

", + ), + ), + ) + + 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="

Devoir commun

", + ), + ), + ) + + 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="

Devoir commun

", + ), + ), + ) + + 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="

Devoir B

" + ), + ), + ) + + 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="

Devoir A

" + ), + ), + ) + + 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 = "

Exercice 1 à 5

\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:
+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:
+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:
+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:
+ Matière : Math + Professeur : M. Dupont + Salle : 204 + + Pour le 10/09/2026 : + Exercice 1 à 5 page 42. + Pour le 10/09/2026 : + Exercice 6 à 10 page 43. +
+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 = ( + "
\n" + " Pour le 10/09/2026:\n" + " Exercice 1 à 5 page 42.\n" + " Pour le 10/09/2026:\n" + " Exercice 6 à 10 page 43.\n" + "
" + ) + 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) diff --git a/tests/unit/test_pronote_client.py b/tests/unit/test_pronote_client.py new file mode 100644 index 0000000..248c6b0 --- /dev/null +++ b/tests/unit/test_pronote_client.py @@ -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