Files
college-infos/tests/integration/test_caldav_sync.py
Antoine Van Elstraete b4b0247919 feat(M7): synchronisation différentielle CalDAV
Implémente la synchronisation des événements Pronote vers un calendrier
CalDAV (Nextcloud) de façon idempotente et sécurisée.

Production :
- sync/serialization.py : sérialisation Lesson/Homework/SchoolEvent vers
  VEVENT, signature sémantique (exclut DTSTAMP/CREATED/LAST-MODIFIED),
  enveloppe VCALENDAR complète avec VERSION:2.0 et PRODID
- sync/caldav.py : passerelle CalDAV isolant caldav>=1.3.0, résolution du
  calendrier via principal().calendars() avec boundary matching, upsert par
  UID (fetch-then-save), exceptions expurgées et __context__ propre, mot de
  passe non stocké en clair, context manager
- sync/planner.py : calcul explicite du CalDAVSyncPlan (add/update/remove
  par comparaison de signatures sémantiques, routage par préfixe d'UID)
- sync/executor.py : exécution du plan avec dry-run (aucune écriture),
  isolation des erreurs par événement, statut FAILED/SKIPPED/SUCCESS
- sync/synchronizer.py : orchestration en trois phases (scan, plan,
  exécution), SKIPPED si CalDAV non configuré
- sync/__init__.py : export synchronize()
- sources/pronote/client.py : normalisation UID via normalize_pronote_uid/
  generate_deterministic_uid (parité avec ical.py)
- config/settings.py : CalDAVSettings durci (url SecretStr, validation
  HTTPS, allow_insecure_http pour localhost, serializer redact_url)

Tests (381 passés, couverture 95.58%) :
- tests/unit/test_sync_serialization.py (21 tests)
- tests/unit/test_caldav_planner.py (16 tests)
- tests/unit/test_caldav_executor.py (18 tests)
- tests/unit/test_caldav_gateway.py (24 tests)
- tests/unit/test_caldav_security.py (18 tests)
- tests/unit/test_uid_equivalence.py (8 tests)
- tests/integration/test_caldav_sync.py (11 tests, faux serveur en mémoire)
- tests/conftest.py : fixtures partagées

Documentation :
- GUIDE_DEV_PYTHON.md §7 : API réelle caldav>=1.3.0, principal().calendars(),
  VCALENDAR complet, upsert par UID, pas d'état local, événements non gérés
  protégés, CalDAVSettings durci (SecretStr, HTTPS, allow_insecure_http)
- TODO.md : M7 coché
- .env.example : CALDAV_ALLOW_INSECURE_HTTP=false

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
Co-authored-by: opencode/tech-writer <tech-writer@agents.invalid>
2026-09-07 09:24:18 +02:00

773 lines
25 KiB
Python

"""Tests d'intégration pour la synchronisation CalDAV.
Ce module teste le flux complet de synchronisation avec un faux serveur
CalDAV en mémoire, en vérifiant l'idempotence, le mode dry-run, les
ajouts/mises à jour/suppressions, et la préservation des événements non gérés.
"""
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING, Any
import pytest
from caldav.lib.error import NotFoundError
from icalendar import Calendar, Event
from pydantic import SecretStr
from pronote_sync.config.settings import AppSettings, CalDAVSettings, Settings
from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent, SchoolEventKind
from pronote_sync.models.homework import Homework
from pronote_sync.models.pronote import PronoteData
from pronote_sync.models.sync import CalDAVSyncStatus
from pronote_sync.sync.synchronizer import synchronize
if TYPE_CHECKING:
pass
class FakeCalendarEvent:
"""Faux événement calendrier pour les tests.
Reproduit le contrat minimal de ``CalendarObjectResource`` utilisé par
la passerelle : contenu brut accessible via ``data`` (lecture/écriture),
enregistrement via ``save()``, composant iCalendar et suppression.
"""
def __init__(
self, ical_text: str, uid: str | None = None, server: FakeCalDAVServer | None = None
):
self._ical_text = ical_text
self._uid = uid
self._server = server
@property
def data(self) -> str:
"""Retourne le contenu iCalendar brut de l'événement.
:return: Texte iCalendar de l'événement.
:rtype: str
"""
return self._ical_text
@data.setter
def data(self, value: str) -> None:
"""Remplace le contenu iCalendar brut de l'événement.
:param value: Nouveau texte iCalendar de l'événement.
"""
self._ical_text = value
def save(self) -> None:
"""Enregistre le contenu courant de l'événement sur le serveur."""
if self._server and self._uid:
self._server._events[self._uid] = self._ical_text
@property
def icalendar_component(self) -> Calendar:
"""Retourne le composant iCalendar de l'événement.
:return: Composant iCalendar parsé.
:rtype: Calendar
"""
return Calendar.from_ical(self._ical_text)
def delete(self) -> None:
"""Supprime l'événement du serveur."""
if self._server and self._uid:
self._server._events.pop(self._uid, None)
class FakeCalendar:
"""Faux calendrier CalDAV pour les tests."""
def __init__(self, server: FakeCalDAVServer):
self._server = server
self.url = "https://caldav.example.com/remote.php/dav/calendars/test-user/pronote-sync/"
def date_search(
self, start: datetime, end: datetime, expand: bool = True
) -> list[FakeCalendarEvent]:
"""Retourne tous les événements comme de faux objets calendrier."""
results: list[FakeCalendarEvent] = []
for uid, ical_text in self._server._events.items():
results.append(FakeCalendarEvent(ical_text, uid=uid, server=self._server))
return results
def get_event_by_uid(self, uid: str) -> FakeCalendarEvent:
"""Retourne un événement par son UID, ou lève NotFoundError s'il est absent."""
if uid in self._server._events:
return FakeCalendarEvent(self._server._events[uid], uid=uid, server=self._server)
raise NotFoundError(f"Event {uid} not found")
def add_event(self, *, ical: str) -> None:
"""Ajoute un événement en extrayant l'UID du texte iCalendar."""
cal = Calendar.from_ical(ical)
for component in cal.walk("VEVENT"):
uid = str(component.get("UID"))
self._server._events[uid] = ical
@property
def icalendar_component(self) -> Calendar:
"""Propriété non utilisée pour le calendrier lui-même."""
raise NotImplementedError
class FakePrincipal:
"""Faux principal CalDAV pour les tests."""
def __init__(self, server: FakeCalDAVServer):
self._server = server
def calendars(self) -> list[FakeCalendar]:
"""Retourne les calendriers du principal.
:return: Liste contenant le faux calendrier cible.
:rtype: list[FakeCalendar]
"""
return [FakeCalendar(self._server)]
class FakeDAVClient:
"""Faux client DAV pour les tests."""
def __init__(self, server: FakeCalDAVServer):
self._server = server
def principal(self) -> FakePrincipal:
"""Retourne un faux principal CalDAV pour la découverte."""
return FakePrincipal(self._server)
class FakeCalDAVServer:
"""Faux serveur CalDAV en mémoire pour les tests d'intégration."""
def __init__(self) -> None:
self._events: dict[str, str] = {} # uid -> ical text
def client_factory(self, **kwargs: Any) -> FakeDAVClient:
"""Retourne un faux client DAV."""
return FakeDAVClient(self)
def get_events(self) -> dict[str, str]:
"""Retourne tous les événements stockés."""
return self._events.copy()
def clear(self) -> None:
"""Efface tous les événements."""
self._events.clear()
@pytest.fixture
def fake_caldav_server() -> FakeCalDAVServer:
"""Fournit un faux serveur CalDAV vide pour les tests."""
return FakeCalDAVServer()
@pytest.fixture
def caldav_settings() -> CalDAVSettings:
"""Configuration CalDAV de test."""
return CalDAVSettings(
url=SecretStr("https://caldav.example.com/remote.php/dav/"),
username="test-user",
password=SecretStr("test-password"),
calendar_path="/pronote-sync/",
)
@pytest.fixture
def app_settings() -> AppSettings:
"""Paramètres d'application de test."""
return AppSettings(
dry_run=False,
log_level="INFO",
sync_past_days=7,
sync_future_days=30,
)
@pytest.fixture
def full_settings(caldav_settings: CalDAVSettings, app_settings: AppSettings) -> Settings:
"""Configuration complète de test."""
return Settings(
caldav=caldav_settings,
app=app_settings,
)
def _create_vevent_text(
uid: str,
summary: str,
start: datetime,
end: datetime,
status: str = "CONFIRMED",
managed: bool = True,
) -> str:
"""Crée un texte iCalendar pour un VEVENT.
:param uid: Identifiant unique de l'événement.
:param summary: Résumé de l'événement.
:param start: Date/heure de début.
:param end: Date/heure de fin.
:param status: Statut de l'événement.
:param managed: Si True, ajoute le marqueur de gestion.
:return: Texte iCalendar de l'événement.
:rtype: str
"""
cal = Calendar()
cal.add("prodid", "-//pronote-sync//test//FR")
cal.add("version", "2.0")
event = Event()
event.add("uid", uid)
event.add("summary", summary)
event.add("dtstart", start)
event.add("dtend", end)
event.add("status", status)
if managed:
event.add("X-PRONOTE-SYNC-MANAGED", "v1")
cal.add_component(event)
return cal.to_ical().decode("utf-8") # type: ignore[no-any-return]
class TestCalDAVSynchronize:
"""Tests d'intégration pour la synchronisation CalDAV."""
def test_full_sync_add(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste l'ajout d'un cours lors d'une synchronisation complète."""
# Données Pronote avec un cours
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
subject="Mathématiques",
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.NORMAL,
group=None,
content=None,
)
],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
)
# Vérifications
assert result.status == CalDAVSyncStatus.SUCCESS
assert result.added == 1
assert result.updated == 0
assert result.removed == 0
assert len(fake_caldav_server.get_events()) == 1
def test_full_sync_idempotent(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste l'idempotence : deux synchronisations identiques."""
# Données Pronote avec un cours
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
subject="Mathématiques",
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.NORMAL,
group=None,
content=None,
)
],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Première synchronisation
result1 = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
)
assert result1.added == 1
# Deuxième synchronisation avec les mêmes données
result2 = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
)
assert result2.status == CalDAVSyncStatus.SKIPPED
assert result2.added == 0
assert result2.updated == 0
assert result2.removed == 0
def test_full_sync_dry_run(
self,
fake_caldav_server: FakeCalDAVServer,
caldav_settings: CalDAVSettings,
) -> None:
"""Teste le mode dry-run : aucune écriture sur le serveur."""
app_settings = AppSettings(
dry_run=True,
log_level="INFO",
sync_past_days=7,
sync_future_days=30,
)
full_settings = Settings(
caldav=caldav_settings,
app=app_settings,
)
# Données Pronote avec un cours
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
subject="Mathématiques",
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.NORMAL,
group=None,
content=None,
)
],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation en mode dry-run
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
)
# Vérifications : le résultat indique un ajout, mais le serveur reste vide
assert result.added == 1
assert len(fake_caldav_server.get_events()) == 0
def test_full_sync_update(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste la mise à jour d'un cours existant."""
# Pré-remplir le serveur avec un événement existant
existing_event = _create_vevent_text(
uid="L-001",
summary="Mathématiques",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
managed=True,
)
fake_caldav_server._events["L-001"] = existing_event
# Données Pronote avec le même cours mais avec un sujet modifié
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
subject="Mathématiques",
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.NORMAL,
group=None,
content=None,
)
],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
)
# Vérifications
assert result.status == CalDAVSyncStatus.SUCCESS
assert result.added == 0
assert result.updated == 1
assert result.removed == 0
def test_full_sync_remove(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste la suppression d'un événement distant non présent localement."""
# Pré-remplir le serveur avec un événement existant
existing_event = _create_vevent_text(
uid="L-001",
summary="Mathématiques",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
managed=True,
)
fake_caldav_server._events["L-001"] = existing_event
# Données Pronote sans ce cours
pronote_data = PronoteData(
lessons=[],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
)
# Vérifications
assert result.status == CalDAVSyncStatus.SUCCESS
assert result.added == 0
assert result.updated == 0
assert result.removed == 1
assert len(fake_caldav_server.get_events()) == 0
def test_cancelled_lesson_preserved(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste qu'un cours annulé est synchronisé avec le statut CANCELLED."""
# Données Pronote avec un cours annulé
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
subject="Mathématiques",
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.CANCELLED,
group=None,
content=None,
)
],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
)
# Vérifications
assert result.status == CalDAVSyncStatus.SUCCESS
assert result.added == 1
assert len(fake_caldav_server.get_events()) == 1
# Vérifie que l'événement a le statut CANCELLED
event_text = list(fake_caldav_server.get_events().values())[0]
assert "STATUS:CANCELLED" in event_text
def test_unmanaged_event_untouched(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste qu'un événement non géré n'est pas supprimé."""
# Pré-remplir le serveur avec un événement NON géré
unmanaged_event = _create_vevent_text(
uid="UNMANAGED-001",
summary="Événement personnel",
start=datetime(2026, 1, 15, 10, 0),
end=datetime(2026, 1, 15, 11, 0),
managed=False, # Non géré
)
fake_caldav_server._events["UNMANAGED-001"] = unmanaged_event
# Données Pronote sans ce cours
pronote_data = PronoteData(
lessons=[],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
)
# Vérifications : l'événement non géré est toujours présent
assert result.status == CalDAVSyncStatus.SKIPPED
assert len(fake_caldav_server.get_events()) == 1
assert "UNMANAGED-001" in fake_caldav_server.get_events()
def test_caldav_not_configured(
self,
fake_caldav_server: FakeCalDAVServer,
) -> None:
"""Teste que la synchronisation est ignorée si CalDAV n'est pas configuré."""
# Configuration sans CalDAV
full_settings = Settings(
caldav=CalDAVSettings(
url=None,
username=None,
password=None,
calendar_path="/pronote-sync/",
),
app=AppSettings(
dry_run=False,
log_level="INFO",
sync_past_days=7,
sync_future_days=30,
),
)
# Données Pronote
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
subject="Mathématiques",
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.NORMAL,
group=None,
content=None,
)
],
homeworks=[],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
)
# Vérifications
assert result.status == CalDAVSyncStatus.SKIPPED
assert result.added == 0
assert result.updated == 0
assert result.removed == 0
def test_homework_sync(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste la synchronisation des devoirs."""
# Données Pronote avec un devoir
pronote_data = PronoteData(
lessons=[],
homeworks=[
Homework(
id="hw-001",
subject="Histoire",
teachers=("Prof Bernard",),
assigned_on=datetime(2026, 1, 15).date(),
due_on=datetime(2026, 1, 20).date(),
text="Lire le chapitre 5",
html="<p>Lire le chapitre 5</p>",
)
],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
)
# Vérifications
assert result.status == CalDAVSyncStatus.SUCCESS
assert result.added == 1
assert len(fake_caldav_server.get_events()) == 1
def test_school_event_sync(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste la synchronisation des événements scolaires."""
# Données Pronote avec un événement scolaire
pronote_data = PronoteData(
lessons=[],
homeworks=[],
school_events=[
SchoolEvent(
kind=SchoolEventKind.HOLIDAY,
label="Vacances de Noël",
from_date=datetime(2026, 12, 20).date(),
to_date=datetime(2027, 1, 5).date(),
)
],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
)
# Vérifications
assert result.status == CalDAVSyncStatus.SUCCESS
assert result.added == 1
assert len(fake_caldav_server.get_events()) == 1
def test_multiple_operations(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Teste une synchronisation avec plusieurs opérations (ajout, mise à jour, suppression)."""
# Pré-remplir le serveur avec des événements existants
existing_lesson = _create_vevent_text(
uid="L-001",
summary="Mathématiques",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
managed=True,
)
existing_homework = _create_vevent_text(
uid="homework-hw-001",
summary="Devoir: Histoire",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 18, 0),
managed=True,
)
existing_unmanaged = _create_vevent_text(
uid="UNMANAGED-001",
summary="Événement personnel",
start=datetime(2026, 1, 15, 10, 0),
end=datetime(2026, 1, 15, 11, 0),
managed=False,
)
fake_caldav_server._events["L-001"] = existing_lesson
fake_caldav_server._events["homework-hw-001"] = existing_homework
fake_caldav_server._events["UNMANAGED-001"] = existing_unmanaged
# Données Pronote :
# - L-001 : modifié (mise à jour)
# - L-002 : nouveau (ajout)
# - hw-001 : supprimé (suppression)
# - hw-002 : nouveau (ajout)
pronote_data = PronoteData(
lessons=[
Lesson(
id="L-001",
start=datetime(2026, 1, 15, 8, 0),
end=datetime(2026, 1, 15, 9, 0),
subject="Physique", # Modifié
teachers=("Prof Dupont",),
rooms=("Salle 101",),
status=LessonStatus.NORMAL,
group=None,
content=None,
),
Lesson(
id="L-002",
start=datetime(2026, 1, 16, 8, 0),
end=datetime(2026, 1, 16, 9, 0),
subject="Français",
teachers=("Prof Martin",),
rooms=("Salle 202",),
status=LessonStatus.NORMAL,
group=None,
content=None,
),
],
homeworks=[
Homework(
id="hw-002",
subject="Mathématiques",
teachers=("Prof Dupont",),
assigned_on=datetime(2026, 1, 15).date(),
due_on=datetime(2026, 1, 20).date(),
text="Exercices page 45",
html="<p>Exercices page 45</p>",
)
],
school_events=[],
messages=[],
target_date=datetime(2026, 1, 15).date(),
generated_at=datetime(2026, 1, 15, 0, 0),
)
# Synchronisation
result = synchronize(
pronote_data=pronote_data,
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
)
# Vérifications
assert result.status == CalDAVSyncStatus.SUCCESS
assert result.added == 2 # L-002 et hw-002
assert result.updated == 1 # L-001
assert result.removed == 1 # hw-001
# Vérifie que l'événement non géré est toujours présent
events = fake_caldav_server.get_events()
assert "UNMANAGED-001" in events
assert len(events) == 4 # L-001 (mis à jour), L-002, hw-002, UNMANAGED-001