Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
75dd78ee1e |
+2
-2
@@ -140,7 +140,7 @@
|
||||
"filename": "GUIDE_DEV_PYTHON.md",
|
||||
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
|
||||
"is_verified": false,
|
||||
"line_number": 5133
|
||||
"line_number": 5125
|
||||
}
|
||||
],
|
||||
"tests/unit/test_caldav_gateway.py": [
|
||||
@@ -185,5 +185,5 @@
|
||||
}
|
||||
]
|
||||
},
|
||||
"generated_at": "2026-09-12T17:18:46Z"
|
||||
"generated_at": "2026-09-12T12:04:07Z"
|
||||
}
|
||||
|
||||
@@ -159,12 +159,9 @@ pronote-sync --dry-run
|
||||
- Après chaque login réussi, les credentials exportées par `pronotepy.export_credentials()` sont
|
||||
persistées dans `.pronote_auth_state.json` (permissions `0600`, format JSON versionné, écriture
|
||||
atomique). Le token rotate à chaque session et peut également être rafraîchi pendant l'exécution
|
||||
(refresh automatique pronotepy après une `PronoteAPIError`). La persistance s'applique après
|
||||
chaque login réussi **et après chaque opération de données réussie** (agenda, devoirs, messages)
|
||||
pour garantir la persistance du token valide ; seules les opérations qui se connectent réellement
|
||||
et récupèrent des données déclenchent la persistance. En mode `qr_token`, `get_informations()`
|
||||
est ignorée (retour immédiat `[]` sans connexion ni verrou) et ne déclenche donc aucune
|
||||
persistance.
|
||||
(refresh automatique pronotepy après une `PronoteAPIError`). Les credentials sont persistées après
|
||||
chaque login réussi **et après chaque opération de données réussie** (agenda, devoirs, messages,
|
||||
informations) pour garantir la persistance du token valide.
|
||||
- Les logins suivants utilisent `pronotepy.token_login(**credentials)` avec le token persisté.
|
||||
- En cas d'échec de `token_login` (token expiré/invalide), une `PronoteAuthRotationError` est levée.
|
||||
Cette erreur se propage sans wrapping à travers `PronoteFetcher` et `fetch_step` jusqu'à
|
||||
@@ -175,11 +172,6 @@ pronote-sync --dry-run
|
||||
- `PronoteAuthRotationError` est re-levée telle quelle (`except PronoteAuthRotationError: raise`)
|
||||
dans toutes les couches d'enveloppement du chemin critique (fetch_agenda, fetch_homework,
|
||||
fetch_step). Ne pas l'attraper avec `except Exception` sans la re-léver d'abord.
|
||||
- En mode `qr_token`, `get_informations()` retourne **inconditionnellement** une liste vide
|
||||
(`[]`) sans connexion, verrou, chargement d'état ni appel réseau, et journalise un message
|
||||
INFO unique : l'endpoint `PageActualites` renvoie une erreur pronotepy 20 sur les instances
|
||||
HubEduConnect testées, provoquant un refresh redondant du token. Ce contournement n'est pas
|
||||
configurable ; aucun état anti-répétition n'est conservé.
|
||||
- Le fichier `.pronote_auth_state.json` ne doit jamais être committé (couvert par `.gitignore`).
|
||||
Son contenu (token vivant) ne doit jamais apparaître dans les logs, les messages d'erreur ou
|
||||
les notifications XMPP.
|
||||
|
||||
+5
-1
@@ -25,13 +25,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
- `PRONOTE_URL` ignoré à cause du double préfixe `env_prefix` (renommage `pronote_url` → `url` dans `PronoteSettings`)
|
||||
- `PRONOTE_ENT` rendu optionnel pour les connexions pronotepy directes
|
||||
- `.env.example` corrigé (`eleve.html` → `parent.html`)
|
||||
- #20/#21 — `get_informations()` ignorée en mode `qr_token` : retourne `[]` immédiatement, sans connexion, verrou ni appel réseau, évitant l'échec systématique de l'endpoint `PageActualites` (erreur pronotepy 20 sur les instances HubEduConnect testées) et le refresh redondant du token associé.
|
||||
|
||||
### Changed
|
||||
|
||||
- Wiki `GuidePronote` enrichi : section "Quand l'ENT est obligatoire" (EduConnect/HubEduConnect), exemple Bordeaux
|
||||
- `AGENTS.md` : ajout de la section §13 "Versionnage et releases"
|
||||
|
||||
### Known Issues
|
||||
|
||||
- #20 — Triple authentification pronotepy (double INIT + refresh) lors d'un run
|
||||
- #21 — Erreur pronotepy 20 « La page a expiré ! (11) » sur `get_informations`
|
||||
|
||||
### Tests
|
||||
|
||||
- 694 tests passés, couverture 94.93%
|
||||
|
||||
+5
-13
@@ -2279,12 +2279,6 @@ méthodes agenda/devoirs ne transforment jamais une erreur en liste vide : elles
|
||||
version expurgée puis lèvent une erreur expurgée avec `from None`. Les méthodes de messages et
|
||||
d'informations sont non critiques et peuvent retourner une liste vide avec un warning.
|
||||
|
||||
En mode `qr_token`, `get_informations()` est ignorée : elle retourne immédiatement `[]` sans
|
||||
connexion, verrou ni appel réseau, et journalise un message INFO unique. L'endpoint
|
||||
`PageActualites` renvoie en effet une erreur pronotepy 20 sur les instances HubEduConnect
|
||||
testées, provoquant un refresh redondant du token. Ce comportement n'est pas configurable ;
|
||||
`get_messages()` n'est pas concernée par ce contournement.
|
||||
|
||||
Les objets renvoyés par `client.homework(start, end)` couvrent une fenêtre. Le résultat destiné à
|
||||
un jour cible est donc filtré explicitement sur `homework.date == target_date`.
|
||||
|
||||
@@ -2296,13 +2290,11 @@ d'authentification et de récupération par un verrou POSIX local non bloquant,
|
||||
`.pronote_auth_state.json.lock`, à côté de `.pronote_auth_state.json`.
|
||||
|
||||
Le verrou couvre l'ensemble du cycle QR/token : chargement de l'état, connexion par token ou
|
||||
enrôlement QR initial, opération de données (agenda, devoirs, messages ; informations hors mode
|
||||
`qr_token`), puis persistance des credentials actualisées. En mode `qr_token`, `get_informations()`
|
||||
est ignorée (retour immédiat `[]` sans connexion ni verrou) : elle n'acquiert pas le verrou et ne
|
||||
déclenche aucune persistance. Une tentative concurrente échoue immédiatement avec une erreur d'état
|
||||
d'authentification expurgée ; elle ne patiente pas et ne relance pas l'authentification. Le contenu
|
||||
du token, le PIN et les autres credentials ne sont jamais inclus dans les logs ni dans ce message
|
||||
d'erreur.
|
||||
enrôlement QR initial, opération de données (agenda, devoirs, messages ou informations), puis
|
||||
persistance des credentials actualisées. Une tentative concurrente échoue immédiatement avec une
|
||||
erreur d'état d'authentification expurgée ; elle ne patiente pas et ne relance pas
|
||||
l'authentification. Le contenu du token, le PIN et les autres credentials ne sont jamais inclus
|
||||
dans les logs ni dans ce message d'erreur.
|
||||
|
||||
Ce mécanisme est un contrat **local** : il coordonne des processus sur le même hôte Linux et un
|
||||
filesystem local. Pour des déploiements conteneurisés, les conteneurs qui partagent le même compte
|
||||
|
||||
@@ -66,6 +66,12 @@ processus. Le mode `PRONOTE_AUTH_MODE=qr_token` est incompatible avec cette gara
|
||||
le refuse avant toute connexion afin de ne pas désynchroniser le token local du token distant.
|
||||
Le dry-run ne remplace pas une vérification des paramètres réellement chargés.
|
||||
|
||||
Si le blog RSS est activé, ses GUID ne sont acquittés qu'après confirmation de
|
||||
l'envoi XMPP. Un refus, une exception, l'absence de canal ou un `--dry-run`
|
||||
laisse donc les articles récupérables à l'exécution suivante ; les en-têtes
|
||||
HTTP associés à ces articles suivent la même règle pour éviter un `304` qui
|
||||
masquerait une livraison non confirmée.
|
||||
|
||||
En mode `PRONOTE_AUTH_MODE=qr_token`, le fichier
|
||||
`.pronote_auth_state.json` et son verrou frère sont créés dans le répertoire
|
||||
de travail du service (par exemple `/var/lib/pronote-sync`) avec le mode
|
||||
|
||||
@@ -230,11 +230,13 @@ class PipelineRunner:
|
||||
data = normalize_step(fetched, generated_at=now)
|
||||
|
||||
try:
|
||||
blog_articles = fetch_blog_step(self._blog_client, self._blog_state)
|
||||
blog_result = fetch_blog_step(self._blog_client, self._blog_state)
|
||||
blog_articles = list(blog_result.articles)
|
||||
except PipelineCriticalError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
self._warn("fetch_blog", self._redact(exc))
|
||||
blog_result = None
|
||||
blog_articles = []
|
||||
|
||||
try:
|
||||
@@ -290,8 +292,11 @@ class PipelineRunner:
|
||||
)
|
||||
if self._channel is not None and not self._dry_run:
|
||||
try:
|
||||
if not send_step(self._channel, message):
|
||||
delivered = send_step(self._channel, message)
|
||||
if not delivered:
|
||||
self._warn("send", "Le canal XMPP a refusé l'envoi")
|
||||
elif blog_result is not None and self._blog_state is not None:
|
||||
self._blog_state.acknowledge(blog_result)
|
||||
except PipelineCriticalError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
|
||||
@@ -2,23 +2,28 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pronote_sync.models.blog import BlogArticle
|
||||
from pronote_sync.sources.blog.result import BlogRSSFetchResult
|
||||
from pronote_sync.sources.blog.rss import BlogRSSClient
|
||||
from pronote_sync.sources.blog.state import BlogRSSState
|
||||
from pronote_sync.utils.redaction import redact_exception
|
||||
|
||||
|
||||
def fetch_blog_step(client: BlogRSSClient | None, state: BlogRSSState | None) -> list[BlogArticle]:
|
||||
"""Récupère les articles RSS nouveaux en conservant l'état du client.
|
||||
def fetch_blog_step(client: BlogRSSClient | None, state: BlogRSSState | None) -> BlogRSSFetchResult:
|
||||
"""Récupère les articles RSS nouveaux sans les acquitter.
|
||||
|
||||
L'état des GUID est acquitté séparément par le pipeline après confirmation
|
||||
de la livraison XMPP. Les en-têtes de cache d'une réponse sans article
|
||||
peuvent être conservés immédiatement, car aucune livraison n'est alors en
|
||||
attente.
|
||||
|
||||
:param client: Client RSS configuré, ou ``None`` lorsque le blog est désactivé.
|
||||
:param state: État de déduplication et de cache HTTP associé au run.
|
||||
:return: Nouveaux articles du blog.
|
||||
:rtype: list[BlogArticle]
|
||||
:return: Résultat de récupération, incluant les métadonnées de cache.
|
||||
:rtype: BlogRSSFetchResult
|
||||
:raises RuntimeError: Si la récupération RSS injectée échoue.
|
||||
"""
|
||||
if client is None or state is None:
|
||||
return []
|
||||
return BlogRSSFetchResult()
|
||||
try:
|
||||
etag, last_modified = state.get_cache_headers()
|
||||
result = client.fetch_and_parse(
|
||||
@@ -26,9 +31,8 @@ def fetch_blog_step(client: BlogRSSClient | None, state: BlogRSSState | None) ->
|
||||
)
|
||||
if result.error is not None:
|
||||
raise RuntimeError(result.error) from None
|
||||
if not result.not_modified:
|
||||
state.add_guids(article.id for article in result.articles)
|
||||
if not result.not_modified and not result.articles:
|
||||
state.update_cache_headers(result.etag, result.last_modified)
|
||||
return list(result.articles)
|
||||
return result
|
||||
except Exception as exc:
|
||||
raise RuntimeError(f"Récupération du blog échouée : {redact_exception(exc)}") from None
|
||||
|
||||
@@ -19,6 +19,7 @@ import logging
|
||||
from collections.abc import Iterable
|
||||
from pathlib import Path
|
||||
|
||||
from pronote_sync.sources.blog.result import BlogRSSFetchResult
|
||||
from pronote_sync.utils.redaction import redact_exception, redact_secrets
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -156,6 +157,22 @@ class BlogRSSState:
|
||||
self._known_guids.update(new_guids)
|
||||
self._save()
|
||||
|
||||
def acknowledge(self, result: BlogRSSFetchResult) -> None:
|
||||
"""Acquitte une récupération RSS après sa livraison confirmée.
|
||||
|
||||
Les GUID et les en-têtes de cache sont enregistrés ensemble afin qu'un
|
||||
article dont la livraison a échoué reste récupérable à l'exécution
|
||||
suivante. Une réponse ``304 Not Modified`` n'a rien à acquitter.
|
||||
|
||||
:param result: Résultat RSS livré avec succès.
|
||||
"""
|
||||
if result.not_modified:
|
||||
return
|
||||
self._known_guids.update(article.id for article in result.articles)
|
||||
self._etag = result.etag
|
||||
self._last_modified = result.last_modified
|
||||
self._save()
|
||||
|
||||
def get_cache_headers(self) -> tuple[str | None, str | None]:
|
||||
"""Renvoie les en-têtes de cache HTTP mémorisés.
|
||||
|
||||
|
||||
@@ -488,23 +488,9 @@ class PronoteClient:
|
||||
Chaque entrée est mappée sur un modèle :class:`Message` de type
|
||||
``SURVEY`` si c'est un sondage, ``INFORMATION`` sinon.
|
||||
|
||||
En mode ``qr_token``, la récupération est ignorée sans connexion ni
|
||||
appel réseau : l'endpoint ``PageActualites`` renvoie une erreur
|
||||
pronotepy 20 sur les instances HubEduConnect testées, provoquant un
|
||||
refresh redondant du token. La méthode retourne alors immédiatement
|
||||
une liste vide et journalise un message INFO unique ; ce comportement
|
||||
n'est pas configurable.
|
||||
|
||||
:return: Liste des informations et sondages ; vide en cas d'erreur ou
|
||||
en mode ``qr_token``.
|
||||
:return: Liste des informations et sondages ; vide en cas d'erreur.
|
||||
:rtype: list[Message]
|
||||
"""
|
||||
if self._settings.auth_mode == "qr_token":
|
||||
logger.info(
|
||||
"Récupération des informations Pronote ignorée : endpoint "
|
||||
"PageActualites indisponible en mode d'authentification qr_token."
|
||||
)
|
||||
return []
|
||||
with self._qr_token_operation_lock():
|
||||
try:
|
||||
client = self._connect()
|
||||
|
||||
@@ -1040,6 +1040,110 @@ def test_runner_blog_success_delivers_articles_into_xmpp_message_external_info(
|
||||
assert xmpp_message.external_info.blog_articles[0].title == "Test Article"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("channel_kind", "dry_run", "should_acknowledge"),
|
||||
[
|
||||
("success", False, True),
|
||||
("false", False, False),
|
||||
("exception", False, False),
|
||||
("none", False, False),
|
||||
("success", True, False),
|
||||
],
|
||||
)
|
||||
def test_runner_acknowledges_blog_only_after_confirmed_xmpp_delivery(
|
||||
pipeline_inputs: tuple[Lesson, Homework],
|
||||
tmp_path: Any,
|
||||
channel_kind: str,
|
||||
dry_run: bool,
|
||||
should_acknowledge: bool,
|
||||
) -> None:
|
||||
"""Les GUID RSS restent rejouables tant que XMPP n'a pas confirmé l'envoi.
|
||||
|
||||
:param pipeline_inputs: Données Pronote de test.
|
||||
:param tmp_path: Répertoire temporaire pour l'état RSS.
|
||||
:param channel_kind: Comportement du canal XMPP simulé.
|
||||
:param dry_run: Active ou non le mode simulation.
|
||||
:param should_acknowledge: Indique si l'état RSS doit être acquitté.
|
||||
"""
|
||||
lesson, homework = pipeline_inputs
|
||||
calls: list[str] = []
|
||||
state_file = tmp_path / "blog-state.json"
|
||||
|
||||
class SuccessfulBlogClient:
|
||||
"""Client RSS renvoyant un article non encore livré."""
|
||||
|
||||
def fetch_and_parse(
|
||||
self,
|
||||
*,
|
||||
known_guids: frozenset[str] | None = None,
|
||||
etag: str | None = None,
|
||||
last_modified: str | None = None,
|
||||
) -> BlogRSSFetchResult:
|
||||
"""Retourne un article et des en-têtes de cache déterministes.
|
||||
|
||||
:param known_guids: GUID déjà connus, ignorés dans ce faux client.
|
||||
:param etag: ETag mémorisé, ignoré dans ce faux client.
|
||||
:param last_modified: Date HTTP mémorisée, ignorée dans ce faux client.
|
||||
:return: Résultat RSS avec un article à livrer.
|
||||
:rtype: BlogRSSFetchResult
|
||||
"""
|
||||
del known_guids, etag, last_modified
|
||||
return BlogRSSFetchResult(
|
||||
articles=(
|
||||
BlogArticle(
|
||||
id="article-to-deliver",
|
||||
title="Article à livrer",
|
||||
url="https://example.com/article-to-deliver",
|
||||
published_at=datetime(2026, 9, 8, 12, 0),
|
||||
updated_at=None,
|
||||
category=None,
|
||||
author=None,
|
||||
content_html="<p>Contenu</p>",
|
||||
content_text="Contenu",
|
||||
),
|
||||
),
|
||||
etag="etag-after-delivery",
|
||||
last_modified="Tue, 08 Sep 2026 12:00:00 GMT",
|
||||
)
|
||||
|
||||
channel: Any
|
||||
if channel_kind == "success":
|
||||
channel = StubChannel(calls)
|
||||
elif channel_kind == "false":
|
||||
channel = FailingChannel()
|
||||
elif channel_kind == "exception":
|
||||
channel = ExceptionalChannel()
|
||||
else:
|
||||
channel = None
|
||||
|
||||
runner = PipelineRunner(
|
||||
settings=Settings(blog=Settings().blog.model_copy(update={"enabled": True})),
|
||||
pronote_fetcher=StubFetcher(calls, lesson, homework),
|
||||
caldav_synchronizer=lambda data, settings: successful_sync_result(),
|
||||
agenda_comparator=cast("AgendaComparator | None", StubComparator(calls)),
|
||||
blog_client=cast("BlogRSSClient | None", SuccessfulBlogClient()),
|
||||
blog_state=BlogRSSState(state_file),
|
||||
channel=channel,
|
||||
dry_run=dry_run,
|
||||
now_provider=lambda: datetime(2026, 9, 8, 7, 0),
|
||||
)
|
||||
|
||||
data, errors = runner.run()
|
||||
|
||||
assert data is not None
|
||||
if should_acknowledge:
|
||||
acknowledged_state = BlogRSSState(state_file)
|
||||
assert acknowledged_state.get_known_guids() == frozenset({"article-to-deliver"})
|
||||
assert acknowledged_state.get_cache_headers() == (
|
||||
"etag-after-delivery",
|
||||
"Tue, 08 Sep 2026 12:00:00 GMT",
|
||||
)
|
||||
else:
|
||||
assert not state_file.exists()
|
||||
if channel_kind in {"false", "exception"}:
|
||||
assert any(error.step == "send" for error in errors)
|
||||
|
||||
|
||||
def test_runner_secret_redaction_in_pipeline_errors(
|
||||
pipeline_inputs: tuple[Lesson, Homework],
|
||||
) -> None:
|
||||
|
||||
@@ -14,11 +14,14 @@ Tous les tests utilisent des fichiers temporaires via la fixture ``tmp_path``.
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from pronote_sync.models.blog import BlogArticle
|
||||
from pronote_sync.sources.blog.result import BlogRSSFetchResult
|
||||
from pronote_sync.sources.blog.state import BlogRSSState
|
||||
|
||||
|
||||
@@ -116,6 +119,38 @@ def test_add_guids_empty_noop(tmp_path: Path) -> None:
|
||||
assert state_file.read_text(encoding="utf-8") == original_content
|
||||
|
||||
|
||||
def test_acknowledge_persists_guids_and_cache_headers_together(tmp_path: Path) -> None:
|
||||
"""Vérifie l'acquittement atomique après une livraison confirmée.
|
||||
|
||||
:param tmp_path: Fixture pytest pour un répertoire temporaire.
|
||||
:return: None
|
||||
"""
|
||||
state_file = tmp_path / "state.json"
|
||||
state = BlogRSSState(state_file)
|
||||
article = BlogArticle(
|
||||
id="guid-1",
|
||||
title="Article",
|
||||
url="https://example.com/article",
|
||||
published_at=datetime(2026, 9, 12, 8, 0, tzinfo=UTC),
|
||||
updated_at=None,
|
||||
category=None,
|
||||
author=None,
|
||||
content_html="<p>Contenu</p>",
|
||||
content_text="Contenu",
|
||||
)
|
||||
|
||||
state.acknowledge(
|
||||
BlogRSSFetchResult(
|
||||
articles=(article,),
|
||||
etag="etag-1",
|
||||
last_modified="Sat, 12 Sep 2026 08:00:00 GMT",
|
||||
)
|
||||
)
|
||||
|
||||
assert state.get_known_guids() == frozenset({"guid-1"})
|
||||
assert state.get_cache_headers() == ("etag-1", "Sat, 12 Sep 2026 08:00:00 GMT")
|
||||
|
||||
|
||||
def test_state_load_persisted_guids(tmp_path: Path) -> None:
|
||||
"""Vérifie que les GUID persistés sont rechargés dans une nouvelle instance.
|
||||
|
||||
|
||||
@@ -621,17 +621,12 @@ def test_get_messages_degraded_on_error(
|
||||
|
||||
|
||||
def test_get_informations_degraded_on_error(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
pronote_settings: PronoteSettings,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||
) -> None:
|
||||
"""Vérifie que get_informations retourne une liste vide en cas d'erreur réseau.
|
||||
|
||||
Assert que le chemin d'erreur retourne toujours [] avec un log ERROR.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param pronote_settings: Paramètres Pronote valides.
|
||||
:param caplog: Fixture pour capturer les logs.
|
||||
:return: None
|
||||
"""
|
||||
mock_client = mocker.MagicMock()
|
||||
@@ -639,254 +634,14 @@ def test_get_informations_degraded_on_error(
|
||||
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||
|
||||
client = PronoteClient(pronote_settings)
|
||||
|
||||
with caplog.at_level(logging.ERROR, logger="pronote_sync.sources.pronote.client"):
|
||||
messages = client.get_informations()
|
||||
messages = client.get_informations()
|
||||
|
||||
assert messages == []
|
||||
# Assert ERROR log is present
|
||||
error_records = [r for r in caplog.records if r.levelno == logging.ERROR]
|
||||
assert len(error_records) >= 1
|
||||
assert any(
|
||||
"Échec de la récupération des informations Pronote" in r.message for r in error_records
|
||||
)
|
||||
|
||||
|
||||
# --- QR code / token authentication tests ---
|
||||
|
||||
|
||||
# --- get_informations qr_token mode guard tests ---
|
||||
|
||||
|
||||
def test_get_informations_skips_in_qr_token_mode(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
) -> None:
|
||||
"""Vérifie que get_informations retourne [] immédiatement en mode qr_token.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:return: None
|
||||
"""
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent="bordeaux",
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
)
|
||||
client = PronoteClient(settings)
|
||||
assert client._client is None
|
||||
|
||||
messages = client.get_informations()
|
||||
|
||||
assert messages == []
|
||||
assert client._client is None
|
||||
|
||||
|
||||
def test_get_informations_no_connect_in_qr_token_mode(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
) -> None:
|
||||
"""Vérifie que _connect n'est jamais appelé en mode qr_token pour get_informations.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:return: None
|
||||
"""
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent="bordeaux",
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
)
|
||||
connect_spy = mocker.spy(PronoteClient, "_connect")
|
||||
|
||||
client = PronoteClient(settings)
|
||||
messages = client.get_informations()
|
||||
|
||||
assert messages == []
|
||||
connect_spy.assert_not_called()
|
||||
|
||||
|
||||
def test_get_informations_no_information_and_surveys_in_qr_token_mode(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
) -> None:
|
||||
"""Vérifie que information_and_surveys n'est jamais appelé en mode qr_token.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:return: None
|
||||
"""
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent="bordeaux",
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
)
|
||||
mock_client = mocker.MagicMock()
|
||||
mock_client.information_and_surveys = mocker.MagicMock()
|
||||
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||
|
||||
client = PronoteClient(settings)
|
||||
messages = client.get_informations()
|
||||
|
||||
assert messages == []
|
||||
mock_client.information_and_surveys.assert_not_called()
|
||||
|
||||
|
||||
def test_get_informations_qr_token_no_side_effects(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Prouve le contrat complet sans effet de bordure du chemin de saut qr_token.
|
||||
|
||||
Avec PronoteAuthState et _qr_token_operation_lock et _persist_credentials
|
||||
mockés, assert que sur le saut : le gestionnaire de contexte de verrou n'est
|
||||
PAS entré, auth-state load() n'est PAS appelé, _persist_credentials() n'est
|
||||
PAS appelé, et l'export des credentials n'est PAS invoqué. Assert aussi que
|
||||
self._client est inchangé.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param caplog: Fixture pour capturer les logs.
|
||||
:return: None
|
||||
"""
|
||||
auth_state = mocker.MagicMock(spec=PronoteAuthState)
|
||||
auth_state.load = mocker.MagicMock()
|
||||
auth_state.lock = mocker.MagicMock()
|
||||
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent="bordeaux",
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
)
|
||||
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
sentinel = MagicMock()
|
||||
client._client = sentinel
|
||||
persist_spy = mocker.spy(client, "_persist_credentials")
|
||||
|
||||
with caplog.at_level(logging.INFO, logger="pronote_sync.sources.pronote.client"):
|
||||
messages = client.get_informations()
|
||||
|
||||
# Assert no side effects
|
||||
assert messages == []
|
||||
assert client._client is sentinel
|
||||
assert isinstance(sentinel, MagicMock)
|
||||
sentinel.export_credentials.assert_not_called()
|
||||
persist_spy.assert_not_called()
|
||||
auth_state.load.assert_not_called()
|
||||
auth_state.lock.assert_not_called()
|
||||
|
||||
# Assert exactly one INFO log record with the exact message
|
||||
info_records = [r for r in caplog.records if r.levelno == logging.INFO]
|
||||
assert len(info_records) == 1
|
||||
assert (
|
||||
info_records[0].message
|
||||
== "Récupération des informations Pronote ignorée : endpoint PageActualites "
|
||||
"indisponible en mode d'authentification qr_token."
|
||||
)
|
||||
# Assert no secret sentinel appears in any log record
|
||||
for record in caplog.records:
|
||||
assert "testpass" not in record.message
|
||||
assert "testuser" not in record.message
|
||||
assert "pronote.example.com" not in record.message
|
||||
|
||||
|
||||
def test_get_informations_logs_info_in_qr_token_mode(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Vérifie que get_informations log un message INFO exact en mode qr_token.
|
||||
|
||||
Assert exactement un enregistrement logging.INFO avec le message exact
|
||||
(inspection de caplog.records, pas seulement caplog.text), et qu'aucune
|
||||
sentinelle de secret n'apparaît.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param caplog: Fixture pour capturer les logs.
|
||||
:return: None
|
||||
"""
|
||||
settings = PronoteSettings(
|
||||
url="https://pronote.example.com",
|
||||
username="testuser",
|
||||
password=SecretStr("testpass"),
|
||||
ent="bordeaux",
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
)
|
||||
|
||||
with caplog.at_level(logging.INFO, logger="pronote_sync.sources.pronote.client"):
|
||||
client = PronoteClient(settings)
|
||||
messages = client.get_informations()
|
||||
|
||||
assert messages == []
|
||||
# Assert exactly one INFO record with the exact message
|
||||
info_records = [r for r in caplog.records if r.levelno == logging.INFO]
|
||||
assert len(info_records) == 1
|
||||
assert (
|
||||
info_records[0].message
|
||||
== "Récupération des informations Pronote ignorée : endpoint PageActualites "
|
||||
"indisponible en mode d'authentification qr_token."
|
||||
)
|
||||
# Assert no secret sentinel appears in any record
|
||||
for record in caplog.records:
|
||||
assert "testpass" not in record.message
|
||||
assert "testuser" not in record.message
|
||||
assert "pronote.example.com" not in record.message
|
||||
|
||||
|
||||
def test_get_informations_unchanged_in_password_mode(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
pronote_settings: PronoteSettings,
|
||||
) -> None:
|
||||
"""Vérifie que get_informations en mode password reste inchangé (régression).
|
||||
|
||||
Assert que _connect() A ÉTÉ appelé et information_and_surveys() A ÉTÉ appelé
|
||||
(en cas de succès), en conservant les assertions de mappage existantes.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param pronote_settings: Paramètres Pronote valides en mode password.
|
||||
: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]
|
||||
|
||||
# Patch _connect to return mock_client and track calls
|
||||
connect_patch = mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||
|
||||
client = PronoteClient(pronote_settings)
|
||||
messages = client.get_informations()
|
||||
|
||||
# Assert _connect() WAS called and information_and_surveys() WAS called
|
||||
connect_patch.assert_called_once()
|
||||
mock_client.information_and_surveys.assert_called_once()
|
||||
|
||||
# Retain existing mapping assertions
|
||||
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_connect_password_mode_unchanged(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
pronote_settings: PronoteSettings,
|
||||
|
||||
Reference in New Issue
Block a user