Antoine Van Elstraete
bfae1ca87f
Corrige les 5 points de l'audit FIXME_M5 : 1. (Bloquant) Sépare transport HTTP et parsing : utilise requests.get() avec timeout explicite et en-têtes conditionnels, puis transmet le contenu à feedparser.parse() — supprime le paramètre inexistant request_timeout qui faisait échouer toute récupération réelle. 2. Rejette les statuts HTTP 4xx/5xx via raise_for_status() avant le parsing. 3. Préserve les validateurs de cache (etag, last_modified) d'entrée sur les chemins d'échec (exception, bozo) au lieu de les écraser à None. 4. Sauvegarde atomique de BlogRSSState : écrit dans un .tmp puis Path.replace() pour éviter la corruption sur interruption. 5. Déduplication normale silencieuse : les GUID déjà connus sont ignorés sans warning ; seuls les doublons intra-flux génèrent un avertissement. Tests : 49 tests (32 client + 17 state) dont 11 nouveaux couvrant transport HTTP réel, statuts 401/404/500, préservation des validateurs, en-têtes conditionnels, doublons intra-flux et sauvegarde atomique. Guide : §5 bis.7.1 aligné avec le nouveau pattern transport/parsing. Co-authored-by: opencode/coder <coder@agents.invalid> Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
351 lines
12 KiB
Python
351 lines
12 KiB
Python
"""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
|