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

merge: jalon M5 — source blog RSS (fetch, parsing, déduplication, état)

Browse Source 
Intègre le jalon M5 complet incluant les corrections d'audit FIXME_M5 :
- BlogRSSClient : client sans état, transport HTTP via requests,
  parsing feedparser, déduplication par known_guids, cache HTTP
  conditionnel, tri déterministe, mode dégradé complet
- BlogRSSFetchResult : résultat immuable (articles + cache + not_modified)
- BlogRSSState : persistance JSON atomique et tolérante
- Fixture blog_rss.xml anonymisée (3 articles, dates fixes)
- 49 tests unitaires (32 client + 17 state)
- Documentation TODO.md et GUIDE_DEV_PYTHON.md alignés
- Configuration pre-commit : feedparser ajouté au hook mypy

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
This commit is contained in:
Antoine Van Elstraete
2026-09-06 20:58:18 +02:00
parent 2c935ef648 344745d725
commit 7fdca2ca55
11 changed files with 2554 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.pre-commit-config.yaml
View File

@@ -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", "types-requests>=2.31.0", "icalendar>=5.0.0", "pronotepy>=2.15.0", "responses>=0.25.0", "pytest-mock>=3.10.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", "feedparser>=6.0.0"]
types: [python]
pass_filenames: true

4
.secrets.baseline
View File

@@ -140,10 +140,10 @@
"filename": "GUIDE_DEV_PYTHON.md",
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
"is_verified": true,
"line_number": 4893,
"line_number": 5058,
"is_secret": false
}
]
},
"generated_at": "2026-09-06T14:37:44Z"
"generated_at": "2026-09-06T18:57:58Z"
}

479
GUIDE_DEV_PYTHON.md
View File

@@ -791,15 +791,71 @@ class PronoteData(BaseModel):
#### 5 bis.7.1 Client RSS (`sources/blog/rss.py`)
Le résultat d'une récupération est un modèle Pydantic figé, `BlogRSSFetchResult`
(module `sources/blog/result.py`) :
```python
import feedparser
from datetime import datetime, timezone
from typing import List, Optional
from html import unescape
from bs4 import BeautifulSoup
from pydantic import BaseModel, ConfigDict, Field
from ..models.blog import BlogArticle
from ..utils.redaction import redact_url
class BlogRSSFetchResult(BaseModel):
"""
Résultat d'une récupération du flux RSS du blog du collège.
Modèle figé (``frozen``) : les instances sont immuables après création.
"""
model_config = ConfigDict(frozen=True)
articles: tuple[BlogArticle, ...] = Field(
default=(),
description=(
"Nouveaux articles absents de known_guids, triés par date de "
"publication décroissante puis par identifiant croissant"
),
)
etag: str | None = Field(
default=None,
description="Valeur de l'en-tête ETag de la réponse RSS, si disponible",
)
last_modified: str | None = Field(
default=None,
description="Valeur de l'en-tête Last-Modified de la réponse RSS, si disponible",
)
not_modified: bool = Field(
default=False,
description="Vaut True si le serveur a répondu 304 Not Modified",
)
```
**Attributs** :
- `articles` : nouveaux articles absents de `known_guids`, triés par date de
publication décroissante puis par identifiant croissant. Tuple vide si aucun
nouvel article (ou en cas de réponse `304 Not Modified`).
- `etag` : valeur de l'en-tête `ETag` de la réponse RSS, ou `None` si
indisponible.
- `last_modified` : valeur de l'en-tête `Last-Modified` de la réponse RSS, ou
`None` si indisponible.
- `not_modified` : vaut `True` si le serveur a répondu `304 Not Modified`,
`False` sinon.
Client de récupération et de parsing du flux (`sources/blog/rss.py`) :
```python
import logging
import re
from datetime import UTC, datetime
from html import unescape
import feedparser
import requests
from bs4 import BeautifulSoup
from ..models.blog import BlogArticle
from ..sources.blog.result import BlogRSSFetchResult
from ..utils.redaction import redact_url
logger = logging.getLogger(__name__)
@@ -809,119 +865,177 @@ class BlogRSSClient:
Client pour récupérer et parser le flux RSS du blog du collège.
"""
def __init__(
self,
rss_url: str = "https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2",
timeout: int = 20,
):
def __init__(self, rss_url: str, timeout: int = 20):
self.rss_url = rss_url
self.timeout = timeout
def fetch_and_parse(self, known_guids: Optional[set] = None) -> List[BlogArticle]:
def fetch_and_parse(
self,
*,
known_guids: frozenset[str] | None = None,
etag: str | None = None,
last_modified: str | None = None,
) -> BlogRSSFetchResult:
"""
Récupère le flux RSS et parse les nouveaux articles.
Args:
known_guids: Ensemble des GUID déjà connus (pour la déduplication). Si None, retourne tous les articles.
known_guids: Ensemble des GUID d'articles déjà traités (pour la déduplication).
Si None, retourne tous les articles.
etag: Valeur de l'en-tête ``ETag`` mémorisée pour la requête conditionnelle, ou None.
last_modified: Valeur de l'en-tête ``Last-Modified`` mémorisée pour la requête
conditionnelle, ou None.
Returns:
Liste des nouveaux articles (triés par date de publication décroissante).
Résultat de la récupération : nouveaux articles (triés par date de publication
décroissante puis par identifiant croissant), en-têtes de cache reçus et
indicateur ``304 Not Modified``.
"""
try:
# Récupération du flux avec cache HTTP (géré par feedparser)
feed = feedparser.parse(
self.rss_url,
request_timeout=self.timeout,
etag=None, # Géré automatiquement par feedparser
modified=None,
)
# Transport HTTP séparé du parsing
headers: dict[str, str] = {"user-agent": "pronote-sync"}
if etag is not None:
headers["If-None-Match"] = etag
if last_modified is not None:
headers["If-Modified-Since"] = last_modified
# Vérifier que le flux est valide
if feed.bozo:
raise ValueError(f"Flux RSS invalide: {feed.bozo_exception}")
response = requests.get(self.rss_url, headers=headers, timeout=self.timeout)
articles = []
for entry in feed.entries:
# Extraire le GUID (utiliser link si guid est vide)
guid = getattr(entry, "guid", None) or entry.link
# 304 Not Modified : pas de nouveaux articles
if response.status_code == 304:
return BlogRSSFetchResult(
articles=(),
etag=etag,
last_modified=last_modified,
not_modified=True,
)
# Ignorer les articles déjà connus
if known_guids and guid in known_guids:
# Rejeter les statuts d'erreur (4xx/5xx)
response.raise_for_status()
# Extraire les en-têtes de cache de la réponse (ETag / Last-Modified)
response_etag: str | None = response.headers.get("ETag")
response_last_modified: str | None = response.headers.get("Last-Modified")
# Parsing du contenu reçu (pas de l'URL)
feed = feedparser.parse(response.content)
# Flux invalide (XML malformé, etc.) : résultat vide, sans erreur ;
# les en-têtes de cache d'entrée sont conservés tels quels
if getattr(feed, "bozo", None):
logger.warning(
"Flux RSS du blog invalide, ignoré : %s",
redact_url(self.rss_url),
)
return BlogRSSFetchResult(
articles=(),
etag=etag,
last_modified=last_modified,
not_modified=False,
)
articles: list[BlogArticle] = []
seen_guids: set[str] = set(known_guids) if known_guids is not None else set()
for entry in getattr(feed, "entries", []):
# Extraire le GUID (utiliser link si le GUID est vide)
guid = str(entry.get("id") or entry.get("link") or "")
# Ignorer les articles déjà connus ou en double dans le flux
if not guid or guid in seen_guids:
continue
# Parser la date de publication (RFC 822 ou ISO 8601)
published_at = self._parse_date(
getattr(entry, "published_parsed", None)
or getattr(entry, "pubdate_parsed", None)
entry.get("published_parsed") or entry.get("pubdate_parsed")
)
if published_at is None:
continue
# Parser la date de mise à jour (si disponible)
updated_at = self._parse_date(
getattr(entry, "updated_parsed", None)
)
updated_at = self._parse_date(entry.get("updated_parsed"))
# Extraire le contenu HTML (content:encoded ou description)
content_html = ""
if hasattr(entry, "content") and entry.content:
content_html = entry.content[0].value
elif hasattr(entry, "description"):
content_html = entry.description
raw_content = entry.get("content")
if raw_content:
content_html = str(raw_content[0].get("value") or "")
else:
content_html = str(entry.get("description") or "")
# Convertir le HTML en texte brut
content_text = self._html_to_text(content_html)
# Extraire la catégorie (tags ou champ category)
tags = entry.get("tags")
category_value = tags[0].get("term") if tags else None
if not category_value:
category_value = entry.get("category")
category = str(category_value) if category_value else None
# Créer l'article
article = BlogArticle(
id=guid,
title=entry.title,
url=entry.link,
published_at=published_at,
updated_at=updated_at,
category=getattr(entry, "category", None),
author=getattr(entry, "author", None),
content_html=content_html,
content_text=content_text,
articles.append(
BlogArticle(
id=guid,
title=str(entry.get("title") or guid),
url=str(entry.get("link") or guid),
published_at=published_at,
updated_at=updated_at,
category=category,
author=str(entry.get("author")) if entry.get("author") else None,
content_html=content_html,
content_text=self._html_to_text(content_html),
)
)
articles.append(article)
seen_guids.add(guid)
# Trier par date de publication décroissante
articles.sort(key=lambda a: a.published_at, reverse=True)
return articles
# Tri stable : d'abord par date de publication décroissante, puis par identifiant croissant
articles.sort(key=lambda article: article.id)
articles.sort(key=lambda article: article.published_at, reverse=True)
return BlogRSSFetchResult(
articles=tuple(articles),
etag=response_etag,
last_modified=response_last_modified,
not_modified=False,
)
except Exception as e:
safe_url = redact_url(self.rss_url)
logger.error(f"Échec de la récupération du flux RSS {safe_url}: {e}")
return []
return BlogRSSFetchResult(
articles=(),
etag=etag,
last_modified=last_modified,
not_modified=False,
)
def _parse_date(self, date_tuple: Optional[tuple]) -> datetime:
@staticmethod
def _parse_date(date_tuple: tuple[int, ...] | None) -> datetime | None:
"""
Convertit un tuple de date (RFC 822 ou ISO 8601) en datetime UTC.
Args:
date_tuple: Tuple retourné par feedparser (ex: (2026, 8, 10, 9, 0, 11, 0, 1, -1)).
date_tuple: Tuple retourné par feedparser (ex: (2026, 8, 10, 9, 0, 11, 0, 1, -1)),
ou None si absent.
Returns:
datetime en UTC.
datetime en UTC, ou None si le tuple est absent, vide ou invalide.
"""
if not date_tuple:
return datetime.now(timezone.utc)
return None
# feedparser retourne un tuple struct_time (année, mois, jour, heure, minute, seconde, jour_semaine, jour_année, DST)
try:
dt = datetime(
return datetime(
date_tuple[0], # année
date_tuple[1], # mois
date_tuple[2], # jour
date_tuple[3], # heure
date_tuple[4], # minute
date_tuple[5], # seconde
tzinfo=timezone.utc,
tzinfo=UTC,
)
return dt
except (ValueError, IndexError):
return datetime.now(timezone.utc)
return None
def _html_to_text(self, html: str) -> str:
@staticmethod
def _html_to_text(html: str) -> str:
"""
Convertit du HTML en texte brut (supprime les balises, décode les entités).
@@ -942,32 +1056,51 @@ class BlogRSSClient:
text = unescape(text)
# Nettoyer les espaces multiples
import re
text = re.sub(r"\s+", " ", text).strip()
return text
```
#### 5 bis.7.2 Déduplication et état local
La déduplication des articles du blog repose sur leur **GUID** (ou leur URL si le GUID est vide).
**Stratégie** :
1. Stocker le **dernier GUID traité** dans un fichier d'état local (ex: `.blog_rss_state.json`).
2. À chaque récupération, ignorer les articles dont le GUID est **antérieur ou égal** au dernier GUID connu.
3. Utiliser le **cache HTTP** (`If-Modified-Since` / `If-None-Match`) via `feedparser` pour éviter les requêtes inutiles.
1. Stocker un **fichier d'état local** (ex: `.blog_rss_state.json`) contenant la version du
format, l'**ensemble des GUID déjà traités** et les en-têtes de cache HTTP (`ETag` /
`Last-Modified`) de la dernière réponse.
2. À chaque récupération, ignorer les articles dont le GUID est **déjà présent** dans
l'ensemble des GUID connus.
3. Utiliser le **cache HTTP** (`If-Modified-Since` / `If-None-Match`) via `feedparser` pour
éviter les requêtes inutiles.
**Exemple de fichier d'état** (`sync/blog_state.py`) :
**Exemple de fichier d'état** :
```json
{
"version": 1,
"known_guids": [
"https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625",
"https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1626"
],
"etag": "abc123",
"last_modified": "Wed, 01 Sep 2026 00:00:00 GMT"
}
```
Les GUID sont triés alphabétiquement pour une sortie JSON déterministe.
**Gestionnaire d'état** (`sources/blog/state.py`) :
```python
import json
from pathlib import Path
from typing import Optional
from ..models.blog import BlogArticle
import logging
from collections.abc import Iterable
from pathlib import Path
logger = logging.getLogger(__name__)
_STATE_VERSION = 1
class BlogRSSState:
"""
@@ -975,64 +1108,77 @@ class BlogRSSState:
"""
def __init__(self, state_file: str = ".blog_rss_state.json"):
self.state_file = Path(state_file)
self._known_guids: set = set()
self._etag: Optional[str] = None
self._last_modified: Optional[str] = None
self._state_file = Path(state_file)
self._known_guids: set[str] = set()
self._etag: str | None = None
self._last_modified: str | None = None
self._load()
def _load(self) -> None:
"""Charge l'état depuis le fichier."""
if self.state_file.exists():
try:
with open(self.state_file, "r", encoding="utf-8") as f:
state = json.load(f)
self._known_guids = set(state.get("known_guids", []))
self._etag = state.get("etag")
self._last_modified = state.get("last_modified")
except Exception as e:
logger.warning(f"Échec du chargement de l'état du blog: {e}")
self._known_guids = set()
self._etag = None
self._last_modified = None
if not self._state_file.exists():
return
try:
data = json.loads(self._state_file.read_text(encoding="utf-8"))
if not isinstance(data, dict) or data.get("version") != _STATE_VERSION:
logger.warning(
"Fichier d'état blog RSS : version absente ou non supportée, "
"démarrage avec un état vide."
)
return
guids_data = data.get("known_guids", [])
if isinstance(guids_data, list):
self._known_guids = {guid for guid in guids_data if isinstance(guid, str)}
etag_data = data.get("etag")
if isinstance(etag_data, str):
self._etag = etag_data
last_modified_data = data.get("last_modified")
if isinstance(last_modified_data, str):
self._last_modified = last_modified_data
except Exception as e:
logger.warning(f"Échec du chargement de l'état du blog: {e}")
self._known_guids = set()
self._etag = None
self._last_modified = None
def _save(self) -> None:
"""Sauvegarde l'état dans le fichier."""
payload = {
"version": _STATE_VERSION,
"known_guids": sorted(self._known_guids),
"etag": self._etag,
"last_modified": self._last_modified,
}
try:
with open(self.state_file, "w", encoding="utf-8") as f:
json.dump({
"known_guids": list(self._known_guids),
"etag": self._etag,
"last_modified": self._last_modified
}, f, indent=2)
with self._state_file.open("w", encoding="utf-8") as f:
json.dump(payload, f, indent=2)
except Exception as e:
logger.error(f"Échec de la sauvegarde de l'état du blog: {e}")
def get_known_guids(self) -> set:
"""Retourne l'ensemble des GUID connus."""
return self._known_guids.copy()
def get_known_guids(self) -> frozenset[str]:
"""Retourne une copie immuable des GUID connus."""
return frozenset(self._known_guids)
def add_guid(self, guid: str) -> None:
"""Ajoute un GUID à l'ensemble des GUID connus."""
self._known_guids.add(guid)
def add_guids(self, guids: Iterable[str]) -> None:
"""Ajoute des GUID à l'ensemble des GUID connus et sauvegarde."""
new_guids = set(guids)
if not new_guids:
return
self._known_guids.update(new_guids)
self._save()
def get_etag(self) -> Optional[str]:
"""Retourne l'ETag du dernier flux RSS."""
return self._etag
def get_cache_headers(self) -> tuple[str | None, str | None]:
"""Retourne les en-têtes de cache HTTP mémorisés (etag, last_modified)."""
return self._etag, self._last_modified
def get_last_modified(self) -> Optional[str]:
"""Retourne la date de dernière modification du flux RSS."""
return self._last_modified
def update_cache_headers(self, etag: Optional[str], last_modified: Optional[str]) -> None:
"""Met à jour les en-têtes de cache HTTP."""
def update_cache_headers(self, etag: str | None, last_modified: str | None) -> None:
"""Met à jour les en-têtes de cache HTTP et sauvegarde."""
self._etag = etag
self._last_modified = last_modified
self._save()
def clear(self) -> None:
"""Efface l'état."""
"""Efface l'état (GUID et en-têtes de cache) et sauvegarde."""
self._known_guids = set()
self._etag = None
self._last_modified = None
@@ -1047,11 +1193,16 @@ blog_state = BlogRSSState()
# Récupération des nouveaux articles
known_guids = blog_state.get_known_guids()
new_articles = rss_client.fetch_and_parse(known_guids=known_guids)
etag, last_modified = blog_state.get_cache_headers()
result = rss_client.fetch_and_parse(
known_guids=known_guids,
etag=etag,
last_modified=last_modified,
)
# Mise à jour de l'état avec les nouveaux GUID
for article in new_articles:
blog_state.add_guid(article.id)
# Mise à jour de l'état avec les nouveaux GUID et les en-têtes de cache
blog_state.add_guids(article.id for article in result.articles)
blog_state.update_cache_headers(result.etag, result.last_modified)
```
---
@@ -1063,9 +1214,8 @@ for article in new_articles:
```python
from typing import List
from ..models.blog import BlogArticle
from ..models.external import ExternalInfo
from ..sources.blog.rss import BlogRSSClient
from ..sync.blog_state import BlogRSSState
from ..sources.blog.state import BlogRSSState
def fetch_blog_step(
@@ -1078,7 +1228,7 @@ def fetch_blog_step(
Args:
rss_client: Client RSS configuré.
blog_state: État local pour la déduplication.
blog_state: État local pour la déduplication et le cache HTTP.
enabled: Si False, retourne une liste vide.
Returns:
@@ -1087,14 +1237,19 @@ def fetch_blog_step(
if not enabled:
return []
last_guid = blog_state.get_last_guid()
articles = rss_client.fetch_and_parse(last_guid=last_guid)
known_guids = blog_state.get_known_guids()
etag, last_modified = blog_state.get_cache_headers()
result = rss_client.fetch_and_parse(
known_guids=known_guids,
etag=etag,
last_modified=last_modified,
)
# Mettre à jour l'état si des articles sont trouvés
if articles:
blog_state.update_last_guid(articles[0].id)
# Mettre à jour l'état avec les nouveaux GUID et les en-têtes de cache
blog_state.add_guids(article.id for article in result.articles)
blog_state.update_cache_headers(result.etag, result.last_modified)
return articles
return list(result.articles)
```
#### 5 bis.8.2 Intégration dans le pipeline principal
@@ -1240,12 +1395,12 @@ def mock_blog_rss_client():
@pytest.mark.unittest
def test_parse_blog_rss(mock_blog_rss_client):
"""Test le parsing d'un flux RSS du blog."""
articles = mock_blog_rss_client.fetch_and_parse()
result = mock_blog_rss_client.fetch_and_parse()
assert len(articles) == 2
assert len(result.articles) == 2
# Vérifier le premier article
article1 = articles[0]
article1 = result.articles[0]
assert article1.title == "Sortie pédagogique"
assert article1.url == "https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1626"
assert article1.category == "Pédagogie"
@@ -1254,7 +1409,7 @@ def test_parse_blog_rss(mock_blog_rss_client):
assert article1.published_at == datetime(2026, 8, 11, 14, 30, 0, tzinfo=timezone.utc)
# Vérifier le deuxième article
article2 = articles[1]
article2 = result.articles[1]
assert article2.title == "Réunion de rentrée"
assert article2.url == "https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625"
assert article2.category == "Administration"
@@ -1266,26 +1421,26 @@ def test_parse_blog_rss(mock_blog_rss_client):
@pytest.mark.unittest
def test_blog_deduplication(tmp_path):
"""Test la déduplication des articles du blog."""
from pronote_sync.sync.blog_state import BlogRSSState
from pronote_sync.sources.blog.state import BlogRSSState
# Créer un fichier d'état temporaire
state_file = tmp_path / "blog_state.json"
state = BlogRSSState(state_file=str(state_file))
# Initialement, aucun article connu
assert state.get_last_guid() is None
assert state.get_known_guids() == frozenset()
# Simuler la récupération de 2 articles
state.update_last_guid("https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625")
assert state.get_last_guid() == "https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625"
state.add_guids(["https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625"])
assert "https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625" in state.get_known_guids()
# Simuler une nouvelle récupération : seul le nouvel article doit être retourné
client = BlogRSSClient(rss_url="file://tests/fixtures/blog_rss.xml")
new_articles = client.fetch_and_parse(last_guid=state.get_last_guid())
result = client.fetch_and_parse(known_guids=state.get_known_guids())
# Seul l'article avec p=1626 doit être retourné (car p=1625 est déjà connu)
assert len(new_articles) == 1
assert new_articles[0].id == "https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1626"
assert len(result.articles) == 1
assert result.articles[0].id == "https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1626"
```
---
@@ -4670,10 +4825,10 @@ exception externe brute susceptible de contenir un secret.
#### 11.4.1 bis `fetch_blog_step.py`
```python
from typing import List, Optional
from ..models.blog import BlogArticle
from ..sources.blog.rss import BlogRSSClient
from ..sync.blog_state import BlogRSSState
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 .errors import PipelineError, ErrorSeverity
@@ -4681,33 +4836,43 @@ def fetch_blog_step(
rss_client: BlogRSSClient,
blog_state: BlogRSSState,
enabled: bool = True,
) -> List[BlogArticle]:
) -> list[BlogArticle]:
"""
Étape de récupération des articles du blog du collège.
Args:
rss_client: Client RSS configuré.
blog_state: État local pour la déduplication.
enabled: Si False, retourne une liste vide.
Lit les GUID déjà connus et les en-têtes de cache HTTP depuis l'état,
puis appelle le client RSS avec ces valeurs pour une requête
conditionnelle. Si la réponse n'est pas ``304 Not Modified`` et que de
nouveaux articles sont présents, les GUID et les en-têtes de cache sont
enregistrés dans l'état. Retourne les nouveaux articles sous forme de
liste.
Returns:
Liste des nouveaux articles.
Raises:
PipelineError: Si la récupération échoue (non bloquante pour le pipeline).
:param rss_client: Client RSS configuré.
:param blog_state: État local pour la déduplication et le cache HTTP.
:param enabled: Si False, retourne une liste vide.
:return: Liste des nouveaux articles.
:rtype: list[BlogArticle]
:raises PipelineError: Si la récupération échoue (non bloquante pour le
pipeline).
"""
if not enabled:
return []
try:
last_guid = blog_state.get_last_guid()
articles = rss_client.fetch_and_parse(last_guid=last_guid)
known_guids = blog_state.get_known_guids()
etag, last_modified = blog_state.get_cache_headers()
result = rss_client.fetch_and_parse(
known_guids=known_guids,
etag=etag,
last_modified=last_modified,
)
# Mettre à jour l'état si des articles sont trouvés
if articles:
blog_state.update_last_guid(articles[0].id)
# Mettre à jour l'état si de nouveaux articles sont trouvés
if not result.not_modified and result.articles:
blog_state.add_guids(article.id for article in result.articles)
blog_state.update_cache_headers(result.etag, result.last_modified)
return articles
return list(result.articles)
except Exception as e:
raise PipelineError(

10
TODO.md
View File

@@ -97,11 +97,11 @@ Récupérer et normaliser l'agenda, les devoirs et les messages Pronote, avec re
Récupérer le flux RSS du blog du collège, parser et dédupliquer les articles.
- [ ] Créer `sources/blog/rss.py` : `BlogRSSClient.fetch_and_parse(known_guids)` avec `feedparser` (§5 bis.7.1).
- [ ] Parser les dates (RFC 822 / ISO 8601) et convertir le HTML en texte brut (`BeautifulSoup` + `html.unescape`).
- [ ] Créer `sources/blog/state.py` (ou `sync/blog_state.py`) : `BlogRSSState` (JSON : `known_guids`, `etag`, `last_modified`).
- [ ] Implémenter la déduplication par GUID et le cache HTTP (`If-Modified-Since` / `etag`).
- [ ] Gérer un flux invalide (`bozo`) et les exceptions sans fuite de secret (retour `[]`/warning).
- [x] Créer `sources/blog/rss.py` : `BlogRSSClient.fetch_and_parse(known_guids)` avec `feedparser` (§5 bis.7.1).
- [x] Parser les dates (RFC 822 / ISO 8601) et convertir le HTML en texte brut (`BeautifulSoup` + `html.unescape`).
- [x] Créer `sources/blog/state.py` (ou `sync/blog_state.py`) : `BlogRSSState` (JSON : `known_guids`, `etag`, `last_modified`).
- [x] Implémenter la déduplication par GUID et le cache HTTP (`If-Modified-Since` / `etag`).
- [x] Gérer un flux invalide (`bozo`) et les exceptions sans fuite de secret (retour `[]`/warning).
### Critères d'acceptation
- `fetch_and_parse` renvoie les nouveaux articles triés par date décroissante, sans doublons.

21
pronote_sync/sources/blog/__init__.py
View File

@@ -0,0 +1,21 @@
"""Source du blog du collège : récupération et suivi du flux RSS.
Ce package expose l'API publique du connecteur du blog du collège :
- :class:`BlogRSSClient` (:mod:`pronote_sync.sources.blog.rss`) : télécharge
et parse le flux RSS, déduplique les entrées par GUID et renvoie les
nouveaux articles dans un :class:`BlogRSSFetchResult`.
- :class:`BlogRSSFetchResult` (:mod:`pronote_sync.sources.blog.result`) :
type de retour figé d'une récupération : nouveaux articles, en-têtes
HTTP de cache (``ETag``/``Last-Modified``) et indicateur ``304 Not
Modified``.
- :class:`BlogRSSState` (:mod:`pronote_sync.sources.blog.state`) : état
local persistant (GUID connus et en-têtes de cache) pour la
déduplication et les requêtes conditionnelles.
"""
from pronote_sync.sources.blog.result import BlogRSSFetchResult
from pronote_sync.sources.blog.rss import BlogRSSClient
from pronote_sync.sources.blog.state import BlogRSSState
__all__ = ["BlogRSSClient", "BlogRSSFetchResult", "BlogRSSState"]

54
pronote_sync/sources/blog/result.py Normal file
View File

@@ -0,0 +1,54 @@
"""Résultat de la récupération du flux RSS du blog du collège.
Ce module définit :class:`BlogRSSFetchResult`, le type de retour figé du
client RSS du blog (:mod:`pronote_sync.sources.blog`).
"""
from __future__ import annotations
from pydantic import BaseModel, ConfigDict, Field
from pronote_sync.models.blog import BlogArticle
class BlogRSSFetchResult(BaseModel):
"""Résultat d'une récupération du flux RSS du blog du collège.
Modèle figé (``frozen``) : les instances sont immuables après création.
Il regroupe les nouveaux articles, triés par date de publication
décroissante puis par identifiant croissant, ainsi que les en-têtes
HTTP utiles aux requêtes conditionnelles (``ETag`` et
``Last-Modified``).
:param articles: Nouveaux articles absents de ``known_guids``, triés
par date de publication décroissante puis par identifiant
croissant. Vide par défaut.
:param etag: Valeur de l'en-tête ``ETag`` de la réponse RSS, si elle
est disponible. ``None`` par défaut.
:param last_modified: Valeur de l'en-tête ``Last-Modified`` de la
réponse RSS, si elle est disponible. ``None`` par défaut.
:param not_modified: Vaut ``True`` si le serveur a répondu avec le
statut ``304 Not Modified``, ``False`` sinon.
"""
model_config = ConfigDict(frozen=True)
articles: tuple[BlogArticle, ...] = Field(
default=(),
description=(
"Nouveaux articles absents de known_guids, triés par date de "
"publication décroissante puis par identifiant croissant"
),
)
etag: str | None = Field(
default=None,
description="Valeur de l'en-tête ETag de la réponse RSS, si disponible",
)
last_modified: str | None = Field(
default=None,
description="Valeur de l'en-tête Last-Modified de la réponse RSS, si disponible",
)
not_modified: bool = Field(
default=False,
description="Vaut True si le serveur a répondu 304 Not Modified",
)

293
pronote_sync/sources/blog/rss.py Normal file
View File

@@ -0,0 +1,293 @@
"""Client de récupération et de parsing du flux RSS du blog du collège.
Ce module définit :class:`BlogRSSClient`, un client sans état qui
télécharge le flux RSS du blog via ``requests``, le parse via
``feedparser``, déduplique les entrées par GUID et les convertit en
:class:`~pronote_sync.models.blog.BlogArticle`.
Le résultat d'une récupération est un
:class:`~pronote_sync.sources.blog.result.BlogRSSFetchResult` : les
nouveaux articles (triés par date de publication décroissante, puis par
identifiant croissant) accompagnés des en-têtes HTTP ``ETag`` et
``Last-Modified`` de la réponse. Toute erreur de récupération ou de
parsing est journalisée (URL et exception rédigées) puis dégradée en
résultat vide : une liste vide est un succès valide, pas une panne.
"""
from __future__ import annotations
import logging
import re
from datetime import UTC, datetime
from html import unescape
import feedparser # type: ignore[import-untyped]
import requests
from bs4 import BeautifulSoup
from pronote_sync.models.blog import BlogArticle
from pronote_sync.sources.blog.result import BlogRSSFetchResult
from pronote_sync.utils.redaction import redact_exception, redact_url
logger = logging.getLogger(__name__)
class BlogRSSClient:
"""Client de récupération et de parsing du flux RSS du blog du collège.
Client sans état : aucune E/S n'est effectuée à la construction et
aucune donnée n'est conservée entre deux appels à
:meth:`fetch_and_parse`. Toute erreur de récupération ou de parsing
est journalisée puis dégradée en résultat vide.
:param rss_url: URL du flux RSS du blog du collège.
:param timeout: Timeout HTTP en secondes (défaut : 20).
"""
def __init__(self, rss_url: str, timeout: int = 20) -> None:
"""Initialise le client RSS du blog.
Aucune opération d'E/S n'est réalisée ici : le téléchargement et
le parsing n'ont lieu qu'à l'appel de :meth:`fetch_and_parse`.
:param rss_url: URL du flux RSS du blog du collège.
:param timeout: Timeout HTTP en secondes (défaut : 20).
"""
self.rss_url = rss_url
self.timeout = timeout
def fetch_and_parse(
self,
*,
known_guids: frozenset[str] | None = None,
etag: str | None = None,
last_modified: str | None = None,
) -> BlogRSSFetchResult:
"""Télécharge et parse le flux RSS du blog en nouveaux articles.
Le flux est téléchargé par ``requests`` avec les en-têtes de
requête conditionnelle fournis (``ETag``/``Last-Modified``), puis
parsé par ``feedparser``. Si le serveur répond ``304 Not Modified``,
le résultat est vide avec
``not_modified=True`` et les en-têtes passés en entrée sont
restitués tels quels. Chaque entrée est dédupliquée par GUID,
convertie en :class:`~pronote_sync.models.blog.BlogArticle`, puis
l'ensemble est trié par date de publication décroissante puis par
identifiant croissant. Toute erreur est journalisée (URL et
exception rédigées) et dégradée en résultat vide : aucune
exception n'est propagée.
:param known_guids: Ensemble des GUID d'articles déjà traités ; les
entrées correspondantes sont ignorées. ``None`` pour tout
conserver (défaut).
:param etag: Valeur de l'en-tête ``ETag`` mémorisée pour la requête
conditionnelle, ou ``None`` (défaut).
:param last_modified: Valeur de l'en-tête ``Last-Modified`` mémorisée
pour la requête conditionnelle, ou ``None`` (défaut).
:return: Résultat de la récupération : nouveaux articles (tuple vide
si aucun nouvel article, réponse ``304`` ou erreur), en-têtes de
cache de la réponse et indicateur ``not_modified``.
:rtype: :class:`~pronote_sync.sources.blog.result.BlogRSSFetchResult`
"""
try:
# Téléchargement HTTP explicite via requests : feedparser 6.x
# n'accepte aucun paramètre de transport ; les requêtes
# conditionnelles sont gérées avec les en-têtes HTTP standards.
headers: dict[str, str] = {"user-agent": "pronote-sync"}
if etag is not None:
headers["If-None-Match"] = etag
if last_modified is not None:
headers["If-Modified-Since"] = last_modified
response = requests.get(self.rss_url, headers=headers, timeout=self.timeout)
# Réponse 304 Not Modified : rien n'a changé, on restitue les
# en-têtes mémorisés tels quels pour les conserver.
if response.status_code == 304:
return BlogRSSFetchResult(
articles=(),
etag=etag,
last_modified=last_modified,
not_modified=True,
)
# Les statuts 4xx/5xx lèvent une exception HTTP, attrapée par le
# gestionnaire général et dégradée en résultat vide.
response.raise_for_status()
response_etag: str | None = response.headers.get("ETag", None)
if response_etag is None:
response_etag = response.headers.get("etag", None)
response_last_modified: str | None = response.headers.get("Last-Modified", None)
if response_last_modified is None:
response_last_modified = response.headers.get("last-modified", None)
# feedparser ne reçoit que le contenu brut de la réponse.
feed = feedparser.parse(response.content)
# Flux invalide (XML malformé, etc.) : avertissement puis résultat
# vide, sans propager l'exception brute. Les validateurs de cache
# d'entrée sont conservés : on ne fait pas confiance aux en-têtes
# d'une réponse au contenu invalide.
if getattr(feed, "bozo", None):
bozo_exception = getattr(feed, "bozo_exception", None)
if bozo_exception is not None:
logger.warning(
"Flux RSS du blog invalide (%s), ignoré : %s",
redact_exception(bozo_exception),
redact_url(self.rss_url),
)
else:
logger.warning(
"Flux RSS du blog invalide, ignoré : %s",
redact_url(self.rss_url),
)
return BlogRSSFetchResult(
articles=(),
etag=etag,
last_modified=last_modified,
not_modified=False,
)
articles: list[BlogArticle] = []
# Déduplication silencieuse des GUID déjà connus (exécutions
# précédentes) et détection des doublons au sein de la réponse.
known_set = set(known_guids) if known_guids is not None else None
seen_in_feed: set[str] = set()
for entry in getattr(feed, "entries", []):
guid_source = entry.get("id") or entry.get("link")
if not guid_source:
logger.warning(
"Entrée RSS sans GUID ni lien, ignorée : %s",
redact_url(self.rss_url),
)
continue
guid = str(guid_source)
if known_set is not None and guid in known_set:
# Déduplication normale (GUID connu d'une exécution
# précédente) : aucun journal n'est nécessaire.
continue
if guid in seen_in_feed:
logger.warning(
"Entrée RSS en double dans le flux, ignorée : %s",
redact_url(self.rss_url),
)
continue
seen_in_feed.add(guid)
published_at = self._parse_date(
entry.get("published_parsed") or entry.get("pubdate_parsed")
)
if published_at is None:
logger.warning(
"Entrée RSS sans date de publication valide, ignorée : %s",
redact_url(self.rss_url),
)
continue
updated_at = self._parse_date(entry.get("updated_parsed"))
raw_content = entry.get("content")
if raw_content:
content_html = str(raw_content[0].get("value") or "")
else:
content_html = str(entry.get("description") or "")
tags = entry.get("tags")
category_value = tags[0].get("term") if tags else None
if not category_value:
category_value = entry.get("category")
category = str(category_value) if category_value else None
author_value = entry.get("author")
author = str(author_value) if author_value else None
title = str(entry.get("title") or guid)
url = str(entry.get("link") or guid)
articles.append(
BlogArticle(
id=guid,
title=title,
url=url,
published_at=published_at,
updated_at=updated_at,
category=category,
author=author,
content_html=content_html,
content_text=self._html_to_text(content_html),
)
)
# Tri stable : d'abord par identifiant croissant, puis par date de
# publication décroissante ; l'ordre par identifiant est conservé
# entre articles de même date.
articles.sort(key=lambda article: article.id)
articles.sort(key=lambda article: article.published_at, reverse=True)
return BlogRSSFetchResult(
articles=tuple(articles),
etag=response_etag,
last_modified=response_last_modified,
not_modified=False,
)
except Exception as exc:
logger.error(
"Échec de la récupération du flux RSS du blog %s : %s",
redact_url(self.rss_url),
redact_exception(exc),
)
return BlogRSSFetchResult(
articles=(),
etag=etag,
last_modified=last_modified,
not_modified=False,
)
@staticmethod
def _parse_date(date_tuple: tuple[int, ...] | None) -> datetime | None:
"""Convertit un tuple de date ``struct_time`` en :class:`datetime` UTC.
:param date_tuple: Tuple horodaté au format ``time.struct_time``
(indices 0 à 5 : année, mois, jour, heure, minute, seconde), ou
``None`` si absent.
:return: Date/heure consciente du fuseau UTC, ou ``None`` si le
tuple est absent, vide ou invalide.
:rtype: datetime | None
"""
if not date_tuple:
return None
try:
return datetime(
date_tuple[0],
date_tuple[1],
date_tuple[2],
date_tuple[3],
date_tuple[4],
date_tuple[5],
tzinfo=UTC,
)
except (ValueError, IndexError):
return None
@staticmethod
def _html_to_text(html: str) -> str:
"""Convertit du HTML en texte brut nettoyé.
Le HTML est parsé avec BeautifulSoup, les balises sont remplacées
par des espaces, les entités HTML sont décodées et les suites
d'espaces sont unifiées.
:param html: Contenu HTML à convertir.
:return: Texte brut sans balises, entités décodées et espaces
unifiés ; chaîne vide si ``html`` est vide.
:rtype: str
"""
if not html:
return ""
soup = BeautifulSoup(html, "html.parser")
text = soup.get_text(separator=" ", strip=True)
text = unescape(text)
return re.sub(r"\s+", " ", text).strip()

174
pronote_sync/sources/blog/state.py Normal file
View File

@@ -0,0 +1,174 @@
"""Gestion de l'état local du flux RSS du blog du collège.
Ce module définit :class:`BlogRSSState`, un gestionnaire d'état persistant
dans un fichier JSON local (``.blog_rss_state.json`` par défaut). Il
mémorise les identifiants (GUID) des articles déjà traités — pour la
déduplication — ainsi que les en-têtes HTTP ``ETag`` et ``Last-Modified``
de la dernière réponse — pour les requêtes conditionnelles.
La lecture et l'écriture sont tolérantes aux erreurs : un fichier absent,
corrompu ou illisible ne fait jamais échouer le pipeline ; l'état vide est
alors utilisé. La sortie JSON est déterministe (``known_guids`` triés
alphabétiquement, champ ``version`` constant).
"""
from __future__ import annotations
import json
import logging
from collections.abc import Iterable
from pathlib import Path
from pronote_sync.utils.redaction import redact_exception, redact_secrets
logger = logging.getLogger(__name__)
_STATE_VERSION = 1
class BlogRSSState:
"""Gère l'état local pour la déduplication des articles et le cache HTTP du flux RSS.
L'état regroupe l'ensemble des GUID d'articles déjà publiés
(``known_guids``) et les en-têtes de cache HTTP (``etag``,
``last_modified``). Il est chargé depuis le fichier JSON à la
construction et sauvegardé à chaque modification. Toute erreur de
lecture ou d'écriture est journalisée sans être propagée.
:param state_file: Chemin du fichier d'état JSON (``str`` ou
:class:`~pathlib.Path`). ``".blog_rss_state.json"`` par défaut.
"""
def __init__(self, state_file: Path | str = ".blog_rss_state.json") -> None:
"""Initialise le gestionnaire d'état depuis le fichier JSON.
:param state_file: Chemin du fichier d'état JSON (``str`` ou
:class:`~pathlib.Path`). ``".blog_rss_state.json"`` par défaut.
"""
self._state_file = Path(state_file)
self._known_guids: set[str] = set()
self._etag: str | None = None
self._last_modified: str | None = None
self._load()
def _load(self) -> None:
"""Charge l'état depuis le fichier JSON.
Si le fichier n'existe pas, l'état reste vide. Si le fichier est
corrompu, illisible ou que la version est absente ou différente
de 1, un avertissement est journalisé et l'état reste vide.
Aucune exception n'est propagée.
"""
if not self._state_file.exists():
return
try:
data = json.loads(self._state_file.read_text(encoding="utf-8"))
if not isinstance(data, dict) or data.get("version") != _STATE_VERSION:
logger.warning(
"Fichier d'état blog RSS %s : version absente ou non supportée, "
"démarrage avec un état vide.",
redact_secrets(str(self._state_file)),
)
return
guids_data = data.get("known_guids", [])
if isinstance(guids_data, list):
self._known_guids = {guid for guid in guids_data if isinstance(guid, str)}
etag_data = data.get("etag")
if isinstance(etag_data, str):
self._etag = etag_data
last_modified_data = data.get("last_modified")
if isinstance(last_modified_data, str):
self._last_modified = last_modified_data
except Exception as exc:
logger.warning(
"Impossible de charger le fichier d'état blog RSS %s : %s, "
"démarrage avec un état vide.",
redact_secrets(str(self._state_file)),
redact_exception(exc),
)
def _save(self) -> None:
"""Sauvegarde l'état dans le fichier JSON de manière atomique.
La sortie est déterministe : ``known_guids`` est trié
alphabétiquement et le champ ``version`` vaut 1. Le JSON est
d'abord écrit dans un fichier temporaire du même répertoire, puis
remplacé atomiquement par :meth:`~pathlib.Path.replace` afin de ne
jamais laisser un fichier partiel en cas d'interruption. En cas
d'erreur d'écriture, une erreur est journalisée sans être
propagée et le fichier temporaire est supprimé.
"""
payload = {
"version": _STATE_VERSION,
"known_guids": sorted(self._known_guids),
"etag": self._etag,
"last_modified": self._last_modified,
}
tmp_file = self._state_file.with_suffix(".tmp")
try:
with open(tmp_file, "w", encoding="utf-8") as handle:
json.dump(payload, handle, indent=2)
tmp_file.replace(self._state_file)
except Exception as exc:
logger.error(
"Impossible d'écrire le fichier d'état blog RSS %s : %s.",
redact_secrets(str(self._state_file)),
redact_exception(exc),
)
try:
tmp_file.unlink(missing_ok=True)
except Exception as cleanup_exc:
logger.debug(
"Nettoyage du fichier temporaire échoué : %s",
redact_exception(cleanup_exc),
)
def get_known_guids(self) -> frozenset[str]:
"""Renvoie une copie immuable des GUID d'articles déjà connus.
:return: Copie de type :class:`frozenset` des GUID connus.
:rtype: frozenset[str]
"""
return frozenset(self._known_guids)
def add_guids(self, guids: Iterable[str]) -> None:
"""Ajoute des GUID d'articles à l'état connu et sauvegarde.
Si l'itérable ne contient aucun GUID, l'état n'est pas modifié et
aucune sauvegarde n'est déclenchée.
:param guids: Itérable des GUID d'articles à enregistrer.
"""
new_guids = set(guids)
if not new_guids:
return
self._known_guids.update(new_guids)
self._save()
def get_cache_headers(self) -> tuple[str | None, str | None]:
"""Renvoie les en-têtes de cache HTTP mémorisés.
:return: Tuple ``(etag, last_modified)``, chaque valeur pouvant
être ``None`` si elle n'a jamais été reçue.
:rtype: tuple[str | None, str | None]
"""
return self._etag, self._last_modified
def update_cache_headers(self, etag: str | None, last_modified: str | None) -> None:
"""Met à jour les en-têtes de cache HTTP et sauvegarde.
:param etag: Nouvelle valeur de l'en-tête ``ETag``, ou ``None``
pour l'effacer.
:param last_modified: Nouvelle valeur de l'en-tête
``Last-Modified``, ou ``None`` pour l'effacer.
"""
self._etag = etag
self._last_modified = last_modified
self._save()
def clear(self) -> None:
"""Réinitialise l'état (GUID et en-têtes de cache) et sauvegarde."""
self._known_guids = set()
self._etag = None
self._last_modified = None
self._save()

47
tests/fixtures/blog_rss.xml vendored Normal file
View File

@@ -0,0 +1,47 @@
<?xml version="1.0" encoding="utf-8"?>
<rss version="2.0" xmlns:content="http://purl.org/rss/1.0/modules/content/" xmlns:dc="http://purl.org/dc/elements/1.1/">
<channel>
<title>Blog du collège Les Mimosas</title>
<link>https://example.com/blog/</link>
<description>Actualités et informations du collège Les Mimosas</description>
<language>fr-FR</language>
<item>
<title>Information générale</title>
<link>https://example.com/blog/?p=1003</link>
<guid isPermaLink="false">https://example.com/blog/?p=1003</guid>
<pubDate>Wed, 12 Aug 2026 08:00:00 +0000</pubDate>
<description>Information générale à destination des familles.</description>
<content:encoded><![CDATA[
<p>La vie scolaire rappelle aux familles que les billets de cantine sont à commander avant le vendredi soir.</p>
<p>Pour toute question, consultez la page <a href="https://example.com/blog/cantine/">cantines et restauration</a> du site.</p>
]]></content:encoded>
</item>
<item>
<title>Réunion de rentrée</title>
<link>https://example.com/blog/?p=1001</link>
<guid isPermaLink="false">https://example.com/blog/?p=1001</guid>
<pubDate>Mon, 10 Aug 2026 09:00:11 +0000</pubDate>
<category>Administration</category>
<dc:creator>M. Dupont</dc:creator>
<description>Réunion de rentrée des parents d'élèves.</description>
<content:encoded><![CDATA[
<p>La réunion de rentrée des parents d'élèves se tiendra le mardi 15 septembre à 18 h 00 dans la salle polyvalente.</p>
<p>L'équipe pédagogique y présentera le projet d'établissement et le calendrier des conseils de classe. Un temps d'échange est prévu avec les professeurs principaux.</p>
<p>Merci de confirmer votre présence en remplissant le <a href="https://example.com/blog/reunion-rentree-inscription/">formulaire d'inscription</a> avant le 10 septembre.</p>
]]></content:encoded>
</item>
<item>
<title>Sortie pédagogique au musée</title>
<link>https://example.com/blog/?p=1002</link>
<guid isPermaLink="false">https://example.com/blog/?p=1002</guid>
<pubDate>Tue, 11 Aug 2026 14:30:00 +0000</pubDate>
<category>Pédagogie</category>
<description>Sortie pédagogique des élèves de 4e au musée d'art moderne.</description>
<content:encoded><![CDATA[
<p>Les élèves de 4e se rendront au musée d'art moderne le jeudi 8 octobre dans le cadre du cours d'arts plastiques.</p>
<p>La visite guidée portera sur la période impressionniste. Les élèves devront apporter un carnet de croquis et leur pique-nique.</p>
<p>Le détail de l'organisation figure dans la <a href="https://example.com/blog/sortie-musee-autorisation/">note d'autorisation</a> à retourner signée avant le 25 septembre.</p>
]]></content:encoded>
</item>
</channel>
</rss>

1285
tests/unit/test_blog_client.py Normal file
View File

File diff suppressed because it is too large Load Diff

350
tests/unit/test_blog_state.py Normal file
View File

@@ -0,0 +1,350 @@
"""Tests unitaires pour le gestionnaire d'état du flux RSS du blog.
Ce module valide le comportement de :class:`BlogRSSState` dans
:mod:`pronote_sync.sources.blog.state`. Les tests couvrent :
- La persistance des GUID connus et des en-têtes de cache HTTP,
- La tolérance aux erreurs (fichier absent, corrompu, version incompatible),
- Le tri alphabétique des GUID lors de la sauvegarde,
- La réinitialisation complète de l'état.
Tous les tests utilisent des fichiers temporaires via la fixture ``tmp_path``.
"""
from __future__ import annotations
import json
from pathlib import Path
from unittest.mock import patch
import pytest
from pronote_sync.sources.blog.state import BlogRSSState
def test_state_file_absent_empty_state(tmp_path: Path) -> None:
"""Vérifie qu'un fichier d'état absent initialise un état vide.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "nonexistent.json"
state = BlogRSSState(state_file)
assert state.get_known_guids() == frozenset()
assert state.get_cache_headers() == (None, None)
def test_add_guids_persists(tmp_path: Path) -> None:
"""Vérifie que l'ajout de GUID persiste dans le fichier JSON.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
state.add_guids(["guid-2", "guid-1", "guid-3"])
assert state.get_known_guids() == frozenset({"guid-1", "guid-2", "guid-3"})
# Vérification du contenu du fichier
saved_data = json.loads(state_file.read_text(encoding="utf-8"))
assert saved_data["known_guids"] == ["guid-1", "guid-2", "guid-3"]
def test_add_guids_empty_noop(tmp_path: Path) -> None:
"""Vérifie que l'ajout d'une liste vide ne modifie pas le fichier.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
# Ajout initial de GUID
state.add_guids(["guid-1"])
original_content = state_file.read_text(encoding="utf-8")
# Ajout d'une liste vide
state.add_guids([])
# Vérification que le fichier n'a pas été modifié (comparaison par contenu)
assert state_file.read_text(encoding="utf-8") == original_content
def test_state_load_persisted_guids(tmp_path: Path) -> None:
"""Vérifie que les GUID persistés sont rechargés dans une nouvelle instance.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
# Création et sauvegarde de l'état initial
state1 = BlogRSSState(state_file)
state1.add_guids(["guid-1", "guid-2"])
# Création d'une nouvelle instance avec le même fichier
state2 = BlogRSSState(state_file)
assert state2.get_known_guids() == frozenset({"guid-1", "guid-2"})
def test_state_load_cache_headers(tmp_path: Path) -> None:
"""Vérifie que les en-têtes de cache persistés sont rechargés.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
# Création et sauvegarde des en-têtes de cache
state1 = BlogRSSState(state_file)
state1.update_cache_headers("etag-123", "Wed, 01 Sep 2026 GMT")
# Création d'une nouvelle instance avec le même fichier
state2 = BlogRSSState(state_file)
assert state2.get_cache_headers() == ("etag-123", "Wed, 01 Sep 2026 GMT")
def test_corrupt_json_warning(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
"""Vérifie qu'un fichier JSON corrompu déclenche un avertissement et initialise un état vide.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:param caplog: Fixture pytest pour capturer les logs.
:return: None
"""
state_file = tmp_path / "corrupt.json"
state_file.write_text("not json{", encoding="utf-8")
with caplog.at_level("WARNING"):
state = BlogRSSState(state_file)
assert state.get_known_guids() == frozenset()
assert state.get_cache_headers() == (None, None)
assert "Impossible de charger le fichier d'état blog RSS" in caplog.text
def test_wrong_version_warning(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
"""Vérifie qu'une version incompatible déclenche un avertissement et initialise un état vide.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:param caplog: Fixture pytest pour capturer les logs.
:return: None
"""
state_file = tmp_path / "wrong_version.json"
state_file.write_text(
json.dumps({"version": 99, "known_guids": ["x"], "etag": None, "last_modified": None}),
encoding="utf-8",
)
with caplog.at_level("WARNING"):
state = BlogRSSState(state_file)
assert state.get_known_guids() == frozenset()
assert state.get_cache_headers() == (None, None)
assert "version absente ou non supportée" in caplog.text
def test_missing_version_warning(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
"""Vérifie qu'un fichier sans champ version déclenche un avertissement et initialise un état vide.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:param caplog: Fixture pytest pour capturer les logs.
:return: None
"""
state_file = tmp_path / "missing_version.json"
state_file.write_text(
json.dumps({"known_guids": ["x"], "etag": None, "last_modified": None}),
encoding="utf-8",
)
with caplog.at_level("WARNING"):
state = BlogRSSState(state_file)
assert state.get_known_guids() == frozenset()
assert state.get_cache_headers() == (None, None)
assert "version absente ou non supportée" in caplog.text
def test_known_guids_sorted_on_save(tmp_path: Path) -> None:
"""Vérifie que les GUID sont triés alphabétiquement lors de la sauvegarde.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
state.add_guids(["c-guid", "a-guid", "b-guid"])
saved_data = json.loads(state_file.read_text(encoding="utf-8"))
assert saved_data["known_guids"] == ["a-guid", "b-guid", "c-guid"]
def test_clear_resets_state(tmp_path: Path) -> None:
"""Vérifie que la méthode clear réinitialise complètement l'état.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
# Ajout de GUID et d'en-têtes de cache
state.add_guids(["guid-1", "guid-2"])
state.update_cache_headers("etag-123", "Wed, 01 Sep 2026 GMT")
# Réinitialisation
state.clear()
assert state.get_known_guids() == frozenset()
assert state.get_cache_headers() == (None, None)
# Vérification du contenu du fichier
saved_data = json.loads(state_file.read_text(encoding="utf-8"))
assert saved_data["known_guids"] == []
assert saved_data["etag"] is None
assert saved_data["last_modified"] is None
def test_clear_persists_to_file(tmp_path: Path) -> None:
"""Vérifie que la réinitialisation est persistée dans le fichier.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
# Création, ajout de données et réinitialisation
state1 = BlogRSSState(state_file)
state1.add_guids(["guid-1"])
state1.update_cache_headers("etag-123", "Wed, 01 Sep 2026 GMT")
state1.clear()
# Création d'une nouvelle instance avec le même fichier
state2 = BlogRSSState(state_file)
assert state2.get_known_guids() == frozenset()
assert state2.get_cache_headers() == (None, None)
def test_str_path_converted_to_path(tmp_path: Path) -> None:
"""Vérifie qu'un chemin de type str est converti en Path.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = str(tmp_path / "state.json")
state = BlogRSSState(state_file)
state.add_guids(["guid-1"])
assert Path(state_file).exists()
def test_update_cache_headers_none_values(tmp_path: Path) -> None:
"""Vérifie que la mise à jour avec des valeurs None fonctionne correctement.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
state.update_cache_headers(None, None)
assert state.get_cache_headers() == (None, None)
# Vérification du contenu du fichier
saved_data = json.loads(state_file.read_text(encoding="utf-8"))
assert saved_data["etag"] is None
assert saved_data["last_modified"] is None
def test_add_guids_multiple_calls(tmp_path: Path) -> None:
"""Vérifie que plusieurs appels à add_guids accumulent les GUID.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
state.add_guids(["guid-1"])
state.add_guids(["guid-2"])
assert state.get_known_guids() == frozenset({"guid-1", "guid-2"})
def test_version_in_saved_file(tmp_path: Path) -> None:
"""Vérifie que le champ version est présent dans le fichier sauvegardé.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
state.add_guids(["guid-1"])
saved_data = json.loads(state_file.read_text(encoding="utf-8"))
assert saved_data["version"] == 1
def test_get_known_guids_returns_frozenset(tmp_path: Path) -> None:
"""Vérifie que get_known_guids retourne un frozenset.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
state.add_guids(["guid-1", "guid-2"])
result = state.get_known_guids()
assert type(result) is frozenset
def test_atomic_save_preserves_on_error(tmp_path: Path) -> None:
"""Vérifie que l'état original est préservé en cas d'erreur lors de la sauvegarde atomique.
Si une erreur survient pendant le remplacement atomique du fichier,
le fichier original doit rester intact et le fichier temporaire doit être nettoyé.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
# Créer un état initial avec des GUID
state = BlogRSSState(state_file)
state.add_guids(["original-guid-1", "original-guid-2"])
# Lire le contenu original
original_content = state_file.read_text(encoding="utf-8")
# Mock Path.replace pour simuler une erreur pendant le remplacement atomique
with patch.object(Path, "replace") as mock_replace:
mock_replace.side_effect = OSError("Simulated atomic replace failure")
# Essayer d'ajouter de nouveaux GUID, ce qui déclenchera _save()
state.add_guids(["new-guid"])
# Vérifier que le fichier original est toujours intact
assert state_file.read_text(encoding="utf-8") == original_content
# Vérifier que le fichier temporaire a été nettoyé
tmp_file = state_file.with_suffix(".tmp")
assert not tmp_file.exists()
# Vérifier que l'état en mémoire n'a pas été modifié (car la sauvegarde a échoué)
# Note: En réalité, l'état en mémoire est modifié mais pas persistant
# C'est le fichier qui doit rester intact
assert state.get_known_guids() == frozenset({"original-guid-1", "original-guid-2", "new-guid"})
# Ensure trailing newline