AntoineVe/college-infos
1
0
Fork 0
Code Issues 16 Pull Requests 1 Actions Packages Projects Releases 3 Wiki Activity

fix: serialize QR token client operations

Browse Source
This commit is contained in:
ChatGPT Codex
2026-09-10 21:24:11 +02:00
parent 5188761209
commit 22a662ab39
2 changed files with 216 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

232
pronote_sync/sources/pronote/client.py
View File

@@ -12,6 +12,8 @@ from __future__ import annotations
import json import json
import logging import logging
from collections.abc import Generator
from contextlib import contextmanager
from datetime import date from datetime import date
from pathlib import Path from pathlib import Path
from typing import Any, Protocol from typing import Any, Protocol
@@ -252,6 +254,26 @@ class PronoteClient:
except Exception as exc: except Exception as exc:
logger.debug("Échec de la persistance des credentials : %s", redact_exception(exc)) logger.debug("Échec de la persistance des credentials : %s", redact_exception(exc))
@contextmanager
def _qr_token_operation_lock(self) -> Generator[None]:
"""Verrouille un cycle d'authentification et de récupération QR/token.
Le verrou englobe le chargement du token, le login, l'opération de
données et la persistance qui suit. Il est volontairement absent du
mode ``password``, qui ne partage pas de fichier d'état de token.
:return: Un gestionnaire de contexte protégeant le cycle QR/token.
:rtype: collections.abc.Generator[None, None, None]
:raises PronoteAuthStateLockError: Si l'état QR/token est déjà utilisé
par une autre opération.
"""
if self._settings.auth_mode != "qr_token" or self._auth_state is None:
yield
return
with self._auth_state.lock():
yield
def _connect_password(self) -> pronotepy.Client: def _connect_password(self) -> pronotepy.Client:
"""Connecte le client ``pronotepy`` en mode ``password``. """Connecte le client ``pronotepy`` en mode ``password``.
@@ -426,37 +448,38 @@ class PronoteClient:
:return: Liste des messages des professeurs ; vide en cas d'erreur. :return: Liste des messages des professeurs ; vide en cas d'erreur.
:rtype: list[Message] :rtype: list[Message]
""" """
try: with self._qr_token_operation_lock():
client = self._connect() try:
messages: list[Message] = [] client = self._connect()
for discussion in client.discussions(): messages: list[Message] = []
for message in discussion.messages: for discussion in client.discussions():
messages.append( for message in discussion.messages:
Message( messages.append(
id=message.id, Message(
type=MessageType.DISCUSSION, id=message.id,
title=discussion.subject, type=MessageType.DISCUSSION,
content=message.content, title=discussion.subject,
author=message.author or "", content=message.content,
date=message.created, author=message.author or "",
read=message.seen, date=message.created,
read=message.seen,
)
) )
) self._persist_credentials()
self._persist_credentials() return messages
return messages except (
except ( pronotepy.PronoteAPIError,
pronotepy.PronoteAPIError, ValueError,
ValueError, requests.RequestException,
requests.RequestException, ConnectionError,
ConnectionError, TimeoutError,
TimeoutError, ) as exc:
) as exc: logger.error(
logger.error( "Échec de la récupération des messages Pronote : %s",
"Échec de la récupération des messages Pronote : %s", redact_exception(exc),
redact_exception(exc), )
) self._persist_credentials()
self._persist_credentials() return []
return []
def get_informations(self) -> list[Message]: def get_informations(self) -> list[Message]:
"""Récupère les informations et sondages Pronote. """Récupère les informations et sondages Pronote.
@@ -467,36 +490,37 @@ class PronoteClient:
:return: Liste des informations et sondages ; vide en cas d'erreur. :return: Liste des informations et sondages ; vide en cas d'erreur.
:rtype: list[Message] :rtype: list[Message]
""" """
try: with self._qr_token_operation_lock():
client = self._connect() try:
messages: list[Message] = [] client = self._connect()
for info in client.information_and_surveys(): messages: list[Message] = []
messages.append( for info in client.information_and_surveys():
Message( messages.append(
id=info.id, Message(
type=MessageType.SURVEY if info.survey else MessageType.INFORMATION, id=info.id,
title=info.title or "", type=MessageType.SURVEY if info.survey else MessageType.INFORMATION,
content=info.content(), title=info.title or "",
author=info.author, content=info.content(),
date=info.creation_date, author=info.author,
read=info.read, date=info.creation_date,
read=info.read,
)
) )
self._persist_credentials()
return messages
except (
pronotepy.PronoteAPIError,
ValueError,
requests.RequestException,
ConnectionError,
TimeoutError,
) as exc:
logger.error(
"Échec de la récupération des informations Pronote : %s",
redact_exception(exc),
) )
self._persist_credentials() self._persist_credentials()
return messages return []
except (
pronotepy.PronoteAPIError,
ValueError,
requests.RequestException,
ConnectionError,
TimeoutError,
) as exc:
logger.error(
"Échec de la récupération des informations Pronote : %s",
redact_exception(exc),
)
self._persist_credentials()
return []
def get_lessons(self, start: date, end: date) -> list[Lesson]: def get_lessons(self, start: date, end: date) -> list[Lesson]:
"""Récupère les cours via ``pronotepy`` (repli iCal). """Récupère les cours via ``pronotepy`` (repli iCal).
@@ -523,37 +547,38 @@ class PronoteClient:
:raises ConnectionError: Si la connexion réseau échoue. :raises ConnectionError: Si la connexion réseau échoue.
:raises TimeoutError: Si la requête réseau expire. :raises TimeoutError: Si la requête réseau expire.
""" """
client = self._connect() with self._qr_token_operation_lock():
lessons: list[Lesson] = [] client = self._connect()
for lesson in client.lessons(start, end): lessons: list[Lesson] = []
content = lesson.content for lesson in client.lessons(start, end):
raw_uid = lesson.id content = lesson.content
if raw_uid: raw_uid = lesson.id
uid = normalize_pronote_uid(raw_uid) if raw_uid:
else: uid = normalize_pronote_uid(raw_uid)
uid = generate_deterministic_uid( else:
start=lesson.start, uid = generate_deterministic_uid(
end=lesson.end, start=lesson.start,
subject=lesson.subject.name if lesson.subject is not None else "", end=lesson.end,
teachers=list(lesson.teacher_names or ()), subject=lesson.subject.name if lesson.subject is not None else "",
rooms=list(lesson.classrooms or ()), teachers=list(lesson.teacher_names or ()),
group=lesson.group_name, rooms=list(lesson.classrooms or ()),
group=lesson.group_name,
)
lessons.append(
Lesson(
id=uid,
start=lesson.start,
end=lesson.end,
subject=lesson.subject.name if lesson.subject is not None else "",
teachers=tuple(lesson.teacher_names or ()),
rooms=tuple(lesson.classrooms or ()),
group=lesson.group_name,
status=(LessonStatus.CANCELLED if lesson.canceled else LessonStatus.NORMAL),
content=content.description if content is not None else None,
)
) )
lessons.append( self._persist_credentials()
Lesson( return lessons
id=uid,
start=lesson.start,
end=lesson.end,
subject=lesson.subject.name if lesson.subject is not None else "",
teachers=tuple(lesson.teacher_names or ()),
rooms=tuple(lesson.classrooms or ()),
group=lesson.group_name,
status=(LessonStatus.CANCELLED if lesson.canceled else LessonStatus.NORMAL),
content=content.description if content is not None else None,
)
)
self._persist_credentials()
return lessons
def get_homeworks(self, start: date, end: date) -> list[Homework]: def get_homeworks(self, start: date, end: date) -> list[Homework]:
"""Récupère les devoirs via ``pronotepy``. """Récupère les devoirs via ``pronotepy``.
@@ -576,19 +601,20 @@ class PronoteClient:
:raises ConnectionError: Si la connexion réseau échoue. :raises ConnectionError: Si la connexion réseau échoue.
:raises TimeoutError: Si la requête réseau expire. :raises TimeoutError: Si la requête réseau expire.
""" """
client = self._connect() with self._qr_token_operation_lock():
homeworks: list[Homework] = [] client = self._connect()
for hw in client.homework(start, end): homeworks: list[Homework] = []
homeworks.append( for hw in client.homework(start, end):
Homework( homeworks.append(
id=hw.id, Homework(
subject=hw.subject.name, id=hw.id,
teachers=(), subject=hw.subject.name,
assigned_on=None, teachers=(),
due_on=hw.date, assigned_on=None,
text=hw.description, due_on=hw.date,
html=hw.description, text=hw.description,
html=hw.description,
)
) )
) self._persist_credentials()
self._persist_credentials() return homeworks
return homeworks

87
tests/unit/test_pronote_client.py
View File

@@ -9,6 +9,8 @@ from __future__ import annotations
import json import json
import logging import logging
from collections.abc import Generator
from contextlib import contextmanager
from datetime import date, datetime from datetime import date, datetime
from pathlib import Path from pathlib import Path
from unittest.mock import MagicMock from unittest.mock import MagicMock
@@ -1203,6 +1205,91 @@ def test_no_raw_secrets_in_logs(
# --- Persistence of credentials after data operations --- # --- Persistence of credentials after data operations ---
def test_qr_token_lock_covers_login_retrieval_and_credential_persistence(
mocker: pytest_mock.MockerFixture,
) -> None:
"""Vérifie que le verrou d'état couvre tout le cycle QR/token des cours.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
events: list[str] = []
credentials = {
"pronote_url": "https://pronote.example.com",
"username": "testuser",
"password": "persisted-token", # pragma: allowlist secret
"uuid": "persisted-uuid",
}
auth_state = mocker.MagicMock(spec=PronoteAuthState)
@contextmanager
def record_lock() -> Generator[None]:
events.append("lock_acquired")
try:
yield
finally:
events.append("lock_released")
def load_credentials() -> dict[str, str]:
"""Simule le chargement des credentials persistés."""
events.append("load")
return credentials
def get_no_lessons(*_: object) -> list[object]:
"""Simule une récupération de cours vide."""
events.append("lessons")
return []
def export_credentials() -> dict[str, str]:
"""Simule l'export des credentials courantes."""
events.append("export")
return credentials
def token_login(**_: object) -> MagicMock:
"""Simule le login par token."""
events.append("token_login")
return mock_client
auth_state.lock.side_effect = record_lock
auth_state.load.side_effect = load_credentials
auth_state.save.side_effect = lambda _: events.append("save")
mock_client = _make_lessons_mock_client(mocker)
mock_client.logged_in = True
mock_client.lessons.side_effect = get_no_lessons
mock_client.export_credentials.side_effect = export_credentials
mocker.patch(
"pronotepy.ParentClient.token_login",
side_effect=token_login,
)
settings = PronoteSettings(
url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent=None,
account_type="parent",
auth_mode="qr_token",
)
lessons = PronoteClient(settings, auth_state=auth_state).get_lessons(
date(2024, 9, 1), date(2024, 9, 30)
)
assert lessons == []
assert events == [
"lock_acquired",
"load",
"token_login",
"export",
"save",
"lessons",
"export",
"save",
"lock_released",
]
def _make_auth_state_mock(mocker: pytest_mock.MockerFixture) -> MagicMock: def _make_auth_state_mock(mocker: pytest_mock.MockerFixture) -> MagicMock:
"""Retourne un mock de PronoteAuthState sans credentials persistés. """Retourne un mock de PronoteAuthState sans credentials persistés.