diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index eaebacb..52dd3c6 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -26,7 +26,7 @@ repos: name: mypy entry: mypy language: python - additional_dependencies: ["mypy>=1.10.0", "pydantic>=2.0.0", "pydantic-settings>=2.0.0", "pytest>=8.0.0", "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 diff --git a/.secrets.baseline b/.secrets.baseline index 4a25695..31437f1 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -140,10 +140,10 @@ "filename": "GUIDE_DEV_PYTHON.md", "hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa", "is_verified": true, - "line_number": 4893, + "line_number": 5043, "is_secret": false } ] }, - "generated_at": "2026-09-06T14:37:44Z" + "generated_at": "2026-09-06T16:41:06Z" } diff --git a/GUIDE_DEV_PYTHON.md b/GUIDE_DEV_PYTHON.md index 73edb64..2ff869d 100644 --- a/GUIDE_DEV_PYTHON.md +++ b/GUIDE_DEV_PYTHON.md @@ -791,15 +791,70 @@ 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 +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 +864,163 @@ 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) + # Récupération du flux avec requête conditionnelle (ETag / Last-Modified) feed = feedparser.parse( self.rss_url, + etag=etag, + modified=last_modified, request_timeout=self.timeout, - etag=None, # Géré automatiquement par feedparser - modified=None, ) - # Vérifier que le flux est valide - if feed.bozo: - raise ValueError(f"Flux RSS invalide: {feed.bozo_exception}") + # Réponse 304 Not Modified : rien n'a changé, on restitue les en-têtes mémorisés + if getattr(feed, "status", None) == 304: + return BlogRSSFetchResult( + articles=(), + etag=etag, + last_modified=last_modified, + not_modified=True, + ) - articles = [] - for entry in feed.entries: - # Extraire le GUID (utiliser link si guid est vide) - guid = getattr(entry, "guid", None) or entry.link + response_etag: str | None = getattr(feed, "etag", None) + response_last_modified: str | None = getattr(feed, "modified", None) - # Ignorer les articles déjà connus - if known_guids and guid in known_guids: + # Flux invalide (erreur HTTP, XML malformé, etc.) : résultat vide, sans erreur + if getattr(feed, "bozo", None): + logger.warning( + "Flux RSS du blog invalide, ignoré : %s", + redact_url(self.rss_url), + ) + return BlogRSSFetchResult( + articles=(), + etag=response_etag, + last_modified=response_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=(), 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 +1041,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 +1093,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 +1178,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 +1199,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 +1213,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 +1222,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 +1380,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 +1394,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 +1406,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 +4810,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 +4821,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( diff --git a/TODO.md b/TODO.md index b606fe5..25dcc09 100644 --- a/TODO.md +++ b/TODO.md @@ -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. diff --git a/pronote_sync/sources/blog/__init__.py b/pronote_sync/sources/blog/__init__.py index e69de29..18c9fcb 100644 --- a/pronote_sync/sources/blog/__init__.py +++ b/pronote_sync/sources/blog/__init__.py @@ -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"] diff --git a/pronote_sync/sources/blog/result.py b/pronote_sync/sources/blog/result.py new file mode 100644 index 0000000..e8f7770 --- /dev/null +++ b/pronote_sync/sources/blog/result.py @@ -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", + ) diff --git a/pronote_sync/sources/blog/rss.py b/pronote_sync/sources/blog/rss.py new file mode 100644 index 0000000..083417a --- /dev/null +++ b/pronote_sync/sources/blog/rss.py @@ -0,0 +1,266 @@ +"""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 ``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] +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 ``feedparser`` avec les en-têtes de + requête conditionnelle fournis (``ETag``/``Last-Modified``). 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: + feed = feedparser.parse( + self.rss_url, + etag=etag, + modified=last_modified, + request_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 getattr(feed, "status", None) == 304: + return BlogRSSFetchResult( + articles=(), + etag=etag, + last_modified=last_modified, + not_modified=True, + ) + + response_etag: str | None = getattr(feed, "etag", None) + response_last_modified: str | None = getattr(feed, "modified", None) + if response_last_modified is None: + headers = getattr(feed, "headers", None) + if headers is not None: + # Les clés des en-têtes sont en minuscules côté feedparser. + response_last_modified = headers.get("last-modified") or None + + # Flux invalide (erreur HTTP, XML malformé, etc.) : avertissement + # puis résultat vide, sans propager l'exception brute. + 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=response_etag, + last_modified=response_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", []): + 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 guid in seen_guids: + logger.warning( + "Entrée RSS déjà traitée ou en double, ignorée : %s", + redact_url(self.rss_url), + ) + continue + + 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), + ) + ) + + seen_guids.add(guid) + + # 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=(), 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() diff --git a/pronote_sync/sources/blog/state.py b/pronote_sync/sources/blog/state.py new file mode 100644 index 0000000..abe6735 --- /dev/null +++ b/pronote_sync/sources/blog/state.py @@ -0,0 +1,161 @@ +"""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. + + La sortie est déterministe : ``known_guids`` est trié + alphabétiquement et le champ ``version`` vaut 1. En cas d'erreur + d'écriture, une erreur est journalisée sans être propagée. + """ + payload = { + "version": _STATE_VERSION, + "known_guids": sorted(self._known_guids), + "etag": self._etag, + "last_modified": self._last_modified, + } + try: + with self._state_file.open("w", encoding="utf-8") as handle: + json.dump(payload, handle, indent=2) + 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), + ) + + 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() diff --git a/tests/fixtures/blog_rss.xml b/tests/fixtures/blog_rss.xml new file mode 100644 index 0000000..f7729e8 --- /dev/null +++ b/tests/fixtures/blog_rss.xml @@ -0,0 +1,47 @@ + + + + Blog du collège Les Mimosas + https://example.com/blog/ + Actualités et informations du collège Les Mimosas + fr-FR + + Information générale + https://example.com/blog/?p=1003 + https://example.com/blog/?p=1003 + Wed, 12 Aug 2026 08:00:00 +0000 + Information générale à destination des familles. + La vie scolaire rappelle aux familles que les billets de cantine sont à commander avant le vendredi soir.

+

Pour toute question, consultez la page cantines et restauration du site.

+ ]]> + + + Réunion de rentrée + https://example.com/blog/?p=1001 + https://example.com/blog/?p=1001 + Mon, 10 Aug 2026 09:00:11 +0000 + Administration + M. Dupont + Réunion de rentrée des parents d'élèves. + La réunion de rentrée des parents d'élèves se tiendra le mardi 15 septembre à 18 h 00 dans la salle polyvalente.

+

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.

+

Merci de confirmer votre présence en remplissant le formulaire d'inscription avant le 10 septembre.

+ ]]> + + + Sortie pédagogique au musée + https://example.com/blog/?p=1002 + https://example.com/blog/?p=1002 + Tue, 11 Aug 2026 14:30:00 +0000 + Pédagogie + Sortie pédagogique des élèves de 4e au musée d'art moderne. + 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.

+

La visite guidée portera sur la période impressionniste. Les élèves devront apporter un carnet de croquis et leur pique-nique.

+

Le détail de l'organisation figure dans la note d'autorisation à retourner signée avant le 25 septembre.

+ ]]> + + + diff --git a/tests/unit/test_blog_client.py b/tests/unit/test_blog_client.py new file mode 100644 index 0000000..2958fb9 --- /dev/null +++ b/tests/unit/test_blog_client.py @@ -0,0 +1,724 @@ +"""Tests unitaires pour le client RSS du blog du collège. + +Ce module vérifie le comportement du client :class:`BlogRSSClient` pour la +récupération et le parsing du flux RSS du blog. Tous les tests sont unitaires +et utilisent des mocks pour éviter tout accès réseau réel. + +Les tests couvrent : +- Le parsing nominal du flux RSS avec déduplication et tri +- La gestion des réponses 304 Not Modified +- La gestion des flux invalides (bozo) +- La gestion des erreurs réseau +- La conversion HTML vers texte +- L'absence de fuite de secrets dans les logs +- Les méthodes statiques de parsing de dates et de conversion HTML +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import feedparser # type: ignore[import-untyped] +import pytest +import pytest_mock + +from pronote_sync.sources.blog.rss import BlogRSSClient + +# --- Fixtures --- + + +@pytest.fixture +def blog_rss_fixture_path() -> Path: + """Chemin vers le fichier fixture RSS du blog. + + :return: Chemin absolu vers le fichier XML de test. + :rtype: Path + """ + return Path(__file__).parent.parent / "fixtures" / "blog_rss.xml" + + +@pytest.fixture +def real_parsed_feed(blog_rss_fixture_path: Path) -> feedparser.FeedParserDict: + """Résultat réel du parsing du fixture RSS par feedparser. + + :param blog_rss_fixture_path: Chemin vers le fichier fixture. + :return: Objet FeedParserDict parsé. + :rtype: feedparser.FeedParserDict + """ + return feedparser.parse(str(blog_rss_fixture_path)) + + +@pytest.fixture +def blog_client(blog_rss_fixture_path: Path) -> BlogRSSClient: + """Instance de BlogRSSClient pointant vers le fixture local. + + :param blog_rss_fixture_path: Chemin vers le fichier fixture. + :return: Instance de BlogRSSClient. + :rtype: BlogRSSClient + """ + return BlogRSSClient(rss_url=f"file://{blog_rss_fixture_path}") + + +# --- Helper functions for mocking --- + + +def make_mock_entry( + entry_id: str | None = None, + link: str | None = None, + title: str = "Test Article", + published_parsed: tuple[int, ...] | None = None, + pubdate_parsed: tuple[int, ...] | None = None, + updated_parsed: tuple[int, ...] | None = None, + content: list[dict[str, Any]] | None = None, + description: str = "", + tags: list[dict[str, Any]] | None = None, + category: str | None = None, + author: str | None = None, +) -> dict[str, Any]: + """Crée une entrée RSS mockée sous forme de dict. + + :param entry_id: GUID de l'entrée. + :param link: URL de l'entrée. + :param title: Titre de l'entrée. + :param published_parsed: Date de publication parsée (struct_time). + :param pubdate_parsed: Date de publication alternative (struct_time). + :param updated_parsed: Date de mise à jour parsée (struct_time). + :param content: Contenu HTML de l'entrée. + :param description: Description de l'entrée. + :param tags: Tags de l'entrée. + :param category: Catégorie de l'entrée. + :param author: Auteur de l'entrée. + :return: Dict représentant une entrée RSS. + """ + entry: dict[str, Any] = { + "title": title, + "description": description, + "author": author, + "category": category, + "tags": tags, + } + if entry_id is not None: + entry["id"] = entry_id + if link is not None: + entry["link"] = link + if published_parsed is not None: + entry["published_parsed"] = published_parsed + if pubdate_parsed is not None: + entry["pubdate_parsed"] = pubdate_parsed + if updated_parsed is not None: + entry["updated_parsed"] = updated_parsed + if content is not None: + entry["content"] = content + return entry + + +def make_mock_feed( + entries: list[dict[str, Any]] | None = None, + status: int = 200, + bozo: int = 0, + etag: str | None = None, + modified: str | None = None, + headers: dict[str, str] | None = None, + bozo_exception: Exception | None = None, +) -> Any: + """Crée un objet FeedParserDict mocké. + + :param entries: Liste des entrées RSS. + :param status: Code de statut HTTP. + :param bozo: Indicateur d'erreur de parsing. + :param etag: Valeur de l'en-tête ETag. + :param modified: Valeur de l'en-tête Last-Modified. + :param headers: En-têtes HTTP de la réponse. + :param bozo_exception: Exception associée à l'erreur de parsing. + :return: Objet FeedParserDict mocké. + """ + # feedparser.FeedParserDict est un dict-like, mais nous utilisons un objet + # dynamique pour simuler les attributs nécessaires + feed = type("FeedParserDict", (), {})() + feed.entries = entries or [] + feed.status = status + feed.bozo = bozo + feed.etag = etag + feed.modified = modified + feed.headers = headers or {} + if bozo_exception: + feed.bozo_exception = bozo_exception + return feed + + +# --- Tests --- + + +def test_fetch_and_parse_nominal( + mocker: pytest_mock.MockerFixture, + real_parsed_feed: feedparser.FeedParserDict, + blog_rss_fixture_path: Path, +) -> None: + """Vérifie le parsing nominal du flux RSS avec 3 articles triés par date. + + Le fixture contient 3 articles dans un ordre non chronologique. + Le résultat doit contenir les 3 articles triés par date décroissante. + + :param mocker: Fixture pytest-mock pour le mocking. + :param real_parsed_feed: Résultat réel du parsing du fixture. + :param blog_rss_fixture_path: Chemin vers le fichier fixture. + :return: None + """ + # Mock feedparser.parse pour retourner le résultat réel du parsing + mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + return_value=real_parsed_feed, + ) + + client = BlogRSSClient(rss_url="https://example.com/blog/feed") + result = client.fetch_and_parse() + + # Vérification du nombre d'articles + assert len(result.articles) == 3 + + # Vérification de l'ordre : p=1003 (12 Aug), p=1002 (11 Aug), p=1001 (10 Aug) + assert result.articles[0].id == "https://example.com/blog/?p=1003" + assert result.articles[1].id == "https://example.com/blog/?p=1002" + assert result.articles[2].id == "https://example.com/blog/?p=1001" + + # Vérification des titres + assert result.articles[0].title == "Information générale" + assert result.articles[1].title == "Sortie pédagogique au musée" + assert result.articles[2].title == "Réunion de rentrée" + + # Vérification des catégories + assert result.articles[0].category is None # Pas de catégorie pour p=1003 + assert result.articles[1].category == "Pédagogie" + assert result.articles[2].category == "Administration" + + # Vérification des auteurs + assert result.articles[0].author is None # Pas d'auteur pour p=1003 + assert result.articles[1].author is None # Pas d'auteur pour p=1002 + assert result.articles[2].author == "M. Dupont" + + # Vérification des dates de publication + assert result.articles[0].published_at == datetime(2026, 8, 12, 8, 0, 0, tzinfo=UTC) + assert result.articles[1].published_at == datetime(2026, 8, 11, 14, 30, 0, tzinfo=UTC) + assert result.articles[2].published_at == datetime(2026, 8, 10, 9, 0, 11, tzinfo=UTC) + + # Vérification que content_text contient des mots attendus + assert "cantine" in result.articles[0].content_text + assert "musée" in result.articles[1].content_text + assert "réunion" in result.articles[2].content_text.lower() + + +def test_fetch_and_parse_deduplication( + mocker: pytest_mock.MockerFixture, + real_parsed_feed: feedparser.FeedParserDict, +) -> None: + """Vérifie la déduplication avec known_guids. + + Avec known_guids contenant p=1001, seuls p=1002 et p=1003 doivent être retournés. + + :param mocker: Fixture pytest-mock pour le mocking. + :param real_parsed_feed: Résultat réel du parsing du fixture. + :return: None + """ + mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + return_value=real_parsed_feed, + ) + + client = BlogRSSClient(rss_url="https://example.com/blog/feed") + result = client.fetch_and_parse(known_guids=frozenset({"https://example.com/blog/?p=1001"})) + + assert len(result.articles) == 2 + # p=1001 doit être exclu + article_ids = {article.id for article in result.articles} + assert "https://example.com/blog/?p=1001" not in article_ids + assert "https://example.com/blog/?p=1002" in article_ids + assert "https://example.com/blog/?p=1003" in article_ids + + +def test_fetch_and_parse_empty_known_guids( + mocker: pytest_mock.MockerFixture, + real_parsed_feed: feedparser.FeedParserDict, +) -> None: + """Vérifie que known_guids vide (frozenset()) retourne tous les articles. + + :param mocker: Fixture pytest-mock pour le mocking. + :param real_parsed_feed: Résultat réel du parsing du fixture. + :return: None + """ + mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + return_value=real_parsed_feed, + ) + + client = BlogRSSClient(rss_url="https://example.com/blog/feed") + result = client.fetch_and_parse(known_guids=frozenset()) + + assert len(result.articles) == 3 + + +def test_fetch_and_parse_known_guids_none( + mocker: pytest_mock.MockerFixture, + real_parsed_feed: feedparser.FeedParserDict, +) -> None: + """Vérifie que known_guids=None retourne tous les articles. + + :param mocker: Fixture pytest-mock pour le mocking. + :param real_parsed_feed: Résultat réel du parsing du fixture. + :return: None + """ + mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + return_value=real_parsed_feed, + ) + + client = BlogRSSClient(rss_url="https://example.com/blog/feed") + result = client.fetch_and_parse(known_guids=None) + + assert len(result.articles) == 3 + + +def test_fetch_and_parse_304_not_modified( + mocker: pytest_mock.MockerFixture, +) -> None: + """Vérifie la gestion de la réponse 304 Not Modified. + + Doit retourner not_modified=True, articles vide, et conserver les en-têtes d'entrée. + + :param mocker: Fixture pytest-mock pour le mocking. + :return: None + """ + mock_feed = make_mock_feed(status=304, etag="old_etag", modified="old_modified") + mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + return_value=mock_feed, + ) + + client = BlogRSSClient(rss_url="https://example.com/blog/feed") + result = client.fetch_and_parse( + etag="input_etag", + last_modified="input_last_modified", + ) + + assert result.not_modified is True + assert result.articles == () + assert result.etag == "input_etag" + assert result.last_modified == "input_last_modified" + + +def test_fetch_and_parse_bozo_invalid_feed( + mocker: pytest_mock.MockerFixture, + caplog: pytest.LogCaptureFixture, +) -> None: + """Vérifie la gestion d'un flux invalide (bozo=1). + + Doit retourner articles vide, not_modified=False, et logger un avertissement. + + :param mocker: Fixture pytest-mock pour le mocking. + :param caplog: Fixture pour capturer les logs. + :return: None + """ + with caplog.at_level(logging.WARNING): + mock_exception = ValueError("Invalid XML") + mock_feed = make_mock_feed( + bozo=1, + bozo_exception=mock_exception, + etag="test_etag", + modified="test_modified", + ) + mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + return_value=mock_feed, + ) + + client = BlogRSSClient(rss_url="https://example.com/blog/feed") + result = client.fetch_and_parse() + + assert result.articles == () + assert result.not_modified is False + assert result.etag == "test_etag" + assert result.last_modified == "test_modified" + + # Vérification du log d'avertissement + assert "Flux RSS du blog invalide" in caplog.text + # L'URL est présente dans le log mais sans secrets (pas de paramètres sensibles) + assert "blog/feed" in caplog.text + + +def test_fetch_and_parse_network_error( + mocker: pytest_mock.MockerFixture, + caplog: pytest.LogCaptureFixture, +) -> None: + """Vérifie la gestion d'une erreur réseau. + + Doit retourner articles vide, not_modified=False, logger une erreur, et ne pas propager l'exception. + + :param mocker: Fixture pytest-mock pour le mocking. + :param caplog: Fixture pour capturer les logs. + :return: None + """ + with caplog.at_level(logging.ERROR): + mock_exception = ConnectionError("Network error") + mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + side_effect=mock_exception, + ) + + client = BlogRSSClient(rss_url="https://example.com/blog/feed") + # Ne doit pas lever d'exception + result = client.fetch_and_parse() + + assert result.articles == () + assert result.not_modified is False + + # Vérification du log d'erreur + assert "Échec de la récupération du flux RSS du blog" in caplog.text + + +def test_fetch_and_parse_no_secret_leak_in_logs( + mocker: pytest_mock.MockerFixture, + caplog: pytest.LogCaptureFixture, +) -> None: + """Vérifie qu'aucun secret ne fuit dans les logs en cas d'erreur. + + :param mocker: Fixture pytest-mock pour le mocking. + :param caplog: Fixture pour capturer les logs. + :return: None + """ + with caplog.at_level(logging.ERROR): + # Exception contenant un secret + secret = "SECRET_TOKEN_123" # pragma: allowlist secret + mock_exception = ValueError(f"Error with token: {secret}") + mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + side_effect=mock_exception, + ) + + client = BlogRSSClient(rss_url=f"https://example.com/blog/feed?token={secret}") + result = client.fetch_and_parse() + + assert result.articles == () + assert result.not_modified is False + + # Vérification qu'aucun secret ne fuit dans les logs + assert secret not in caplog.text + + +def test_fetch_and_parse_entry_without_date( + mocker: pytest_mock.MockerFixture, + caplog: pytest.LogCaptureFixture, +) -> None: + """Vérifie qu'une entrée sans date de publication est ignorée. + + :param mocker: Fixture pytest-mock pour le mocking. + :param caplog: Fixture pour capturer les logs. + :return: None + """ + with caplog.at_level(logging.WARNING): + entry = make_mock_entry( + entry_id="test-id", + link="https://example.com/test", + title="Test Article", + published_parsed=None, + pubdate_parsed=None, + ) + mock_feed = make_mock_feed(entries=[entry]) + mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + return_value=mock_feed, + ) + + client = BlogRSSClient(rss_url="https://example.com/blog/feed") + result = client.fetch_and_parse() + + assert len(result.articles) == 0 + assert "Entrée RSS sans date de publication valide" in caplog.text + + +def test_fetch_and_parse_entry_without_guid( + mocker: pytest_mock.MockerFixture, + caplog: pytest.LogCaptureFixture, +) -> None: + """Vérifie qu'une entrée sans GUID ni lien est ignorée. + + :param mocker: Fixture pytest-mock pour le mocking. + :param caplog: Fixture pour capturer les logs. + :return: None + """ + with caplog.at_level(logging.WARNING): + entry = make_mock_entry( + entry_id=None, + link=None, + title="Test Article", + published_parsed=(2026, 8, 10, 9, 0, 11, 0, 222, 0), + ) + mock_feed = make_mock_feed(entries=[entry]) + mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + return_value=mock_feed, + ) + + client = BlogRSSClient(rss_url="https://example.com/blog/feed") + result = client.fetch_and_parse() + + assert len(result.articles) == 0 + assert "Entrée RSS sans GUID ni lien" in caplog.text + + +def test_fetch_and_parse_html_to_text_conversion( + mocker: pytest_mock.MockerFixture, + real_parsed_feed: feedparser.FeedParserDict, +) -> None: + """Vérifie que content_text ne contient pas de balises HTML. + + :param mocker: Fixture pytest-mock pour le mocking. + :param real_parsed_feed: Résultat réel du parsing du fixture. + :return: None + """ + mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + return_value=real_parsed_feed, + ) + + client = BlogRSSClient(rss_url="https://example.com/blog/feed") + result = client.fetch_and_parse() + + for article in result.articles: + # Vérification qu'il n'y a pas de balises HTML dans content_text + assert "

" not in article.content_text + assert "

" not in article.content_text + assert "" not in article.content_text + assert " 0 + + +def test_fetch_and_parse_etag_last_modified_returned( + mocker: pytest_mock.MockerFixture, +) -> None: + """Vérifie que les en-têtes ETag et Last-Modified sont retournés. + + :param mocker: Fixture pytest-mock pour le mocking. + :return: None + """ + mock_feed = make_mock_feed( + entries=[], + etag="abc123", + modified="Wed, 01 Sep 2026 00:00:00 GMT", + ) + mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + return_value=mock_feed, + ) + + client = BlogRSSClient(rss_url="https://example.com/blog/feed") + result = client.fetch_and_parse() + + assert result.etag == "abc123" + assert result.last_modified == "Wed, 01 Sep 2026 00:00:00 GMT" + assert result.not_modified is False + + +def test_fetch_and_parse_passes_etag_to_feedparser( + mocker: pytest_mock.MockerFixture, +) -> None: + """Vérifie que les paramètres etag et last_modified sont passés à feedparser.parse. + + :param mocker: Fixture pytest-mock pour le mocking. + :return: None + """ + mock_parse = mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + return_value=make_mock_feed(), + ) + + client = BlogRSSClient(rss_url="https://example.com/blog/feed") + client.fetch_and_parse( + etag="abc123", + last_modified="Wed, 01 Sep 2026 00:00:00 GMT", + ) + + # Vérification que feedparser.parse a été appelé avec les bons paramètres + mock_parse.assert_called_once() + call_args = mock_parse.call_args + assert call_args[1]["etag"] == "abc123" + assert call_args[1]["modified"] == "Wed, 01 Sep 2026 00:00:00 GMT" + + +def test_parse_date_valid() -> None: + """Vérifie le parsing d'une date valide. + + :return: None + """ + date_tuple = (2026, 8, 10, 9, 0, 11, 0, 222, 0) + result = BlogRSSClient._parse_date(date_tuple) + + expected = datetime(2026, 8, 10, 9, 0, 11, tzinfo=UTC) + assert result == expected + + +def test_parse_date_none() -> None: + """Vérifie le parsing d'une date None. + + :return: None + """ + result = BlogRSSClient._parse_date(None) + assert result is None + + +def test_parse_date_invalid() -> None: + """Vérifie le parsing d'une date invalide (tuple trop court). + + :return: None + """ + result = BlogRSSClient._parse_date((2026,)) + assert result is None + + +def test_html_to_text_basic() -> None: + """Vérifie la conversion HTML vers texte de base. + + :return: None + """ + html = "

Hello world

" + result = BlogRSSClient._html_to_text(html) + + assert result == "Hello world" + + +def test_html_to_text_empty() -> None: + """Vérifie la conversion HTML vers texte avec une chaîne vide. + + :return: None + """ + result = BlogRSSClient._html_to_text("") + assert result == "" + + +def test_html_to_text_entities() -> None: + """Vérifie la conversion HTML vers texte avec des entités HTML. + + :return: None + """ + html = "

Café & croissant

" + result = BlogRSSClient._html_to_text(html) + + assert result == "Café & croissant" + + +def test_fetch_and_parse_sort_deterministic( + mocker: pytest_mock.MockerFixture, + real_parsed_feed: feedparser.FeedParserDict, +) -> None: + """Vérifie que le tri des articles est déterministe. + + Deux appels successifs doivent retourner les articles dans le même ordre. + + :param mocker: Fixture pytest-mock pour le mocking. + :param real_parsed_feed: Résultat réel du parsing du fixture. + :return: None + """ + mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + return_value=real_parsed_feed, + ) + + client = BlogRSSClient(rss_url="https://example.com/blog/feed") + + result1 = client.fetch_and_parse() + result2 = client.fetch_and_parse() + + # Vérification que les deux résultats sont identiques + assert len(result1.articles) == len(result2.articles) + for article1, article2 in zip(result1.articles, result2.articles, strict=True): + assert article1.id == article2.id + assert article1.title == article2.title + assert article1.published_at == article2.published_at + + # Vérification de l'ordre : p=1003 (Aug 12), p=1002 (Aug 11), p=1001 (Aug 10) + assert result1.articles[0].id == "https://example.com/blog/?p=1003" + assert result1.articles[1].id == "https://example.com/blog/?p=1002" + assert result1.articles[2].id == "https://example.com/blog/?p=1001" + + +def test_fetch_and_parse_duplicate_guid_in_feed( + mocker: pytest_mock.MockerFixture, +) -> None: + """Vérifie la déduplication des entrées avec le même GUID dans un même flux. + + Deux entrées avec le même GUID (même id et même link) doivent donner un seul article. + + :param mocker: Fixture pytest-mock pour le mocking. + :return: None + """ + # Création de deux entrées avec le même GUID + entry1 = make_mock_entry( + entry_id="duplicate-guid", + link="https://example.com/duplicate", + title="First Article", + published_parsed=(2026, 8, 10, 9, 0, 0, 0, 222, 0), + ) + entry2 = make_mock_entry( + entry_id="duplicate-guid", + link="https://example.com/duplicate", + title="Second Article", + published_parsed=(2026, 8, 11, 10, 0, 0, 0, 223, 0), + ) + + mock_feed = make_mock_feed(entries=[entry1, entry2]) + mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + return_value=mock_feed, + ) + + client = BlogRSSClient(rss_url="https://example.com/blog/feed") + result = client.fetch_and_parse(known_guids=None) + + # Doit retourner un seul article (le second duplicata est ignoré) + assert len(result.articles) == 1 + # Le premier article doit être conservé (celui avec la date la plus ancienne) + assert result.articles[0].id == "duplicate-guid" + assert result.articles[0].title == "First Article" + + +def test_fetch_and_parse_sort_tied_dates_by_id( + mocker: pytest_mock.MockerFixture, +) -> None: + """Vérifie le tri secondaire par identifiant croissant pour les articles de même date. + + Deux articles avec la même date de publication doivent être triés par id croissant. + + :param mocker: Fixture pytest-mock pour le mocking. + :return: None + """ + # Création de deux entrées avec la même date mais des id différents + entry_a = make_mock_entry( + entry_id="guid-a", + link="https://example.com/a", + title="Article A", + published_parsed=(2026, 8, 10, 9, 0, 0, 0, 222, 0), + ) + entry_b = make_mock_entry( + entry_id="guid-b", + link="https://example.com/b", + title="Article B", + published_parsed=(2026, 8, 10, 9, 0, 0, 0, 222, 0), + ) + + mock_feed = make_mock_feed(entries=[entry_b, entry_a]) # Ordre initial : b, a + mocker.patch( + "pronote_sync.sources.blog.rss.feedparser.parse", + return_value=mock_feed, + ) + + client = BlogRSSClient(rss_url="https://example.com/blog/feed") + result = client.fetch_and_parse() + + # Doit retourner les deux articles triés par id croissant + assert len(result.articles) == 2 + assert result.articles[0].id == "guid-a" # a doit venir avant b + assert result.articles[1].id == "guid-b" diff --git a/tests/unit/test_blog_state.py b/tests/unit/test_blog_state.py new file mode 100644 index 0000000..cdb5f3c --- /dev/null +++ b/tests/unit/test_blog_state.py @@ -0,0 +1,311 @@ +"""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 + +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 + + +# Ensure trailing newline