Compare commits

..
Author SHA1 Message Date
Codex 6575a5f60b fix(caldav): rendre les devoirs transparents 2026-09-12 15:43:12 +02:00
6 changed files with 119 additions and 75 deletions
+4 -11
View File
@@ -105,17 +105,10 @@ sudo systemctl start pronote-sync.service
sudo systemctl status pronote-sync.service sudo systemctl status pronote-sync.service
``` ```
La CLI expose un contrat de sortie stable : `0` signifie une exécution complète, Une exécution en échec laisse l'unité `pronote-sync.service` en état `failed`.
`2` une exécution dégradée (les données Pronote sont disponibles mais une étape La supervision de l'hôte doit donc déclencher une alerte sur cet état ou sur un
optionnelle, CalDAV ou XMPP a échoué), et `1` un échec critique. Tout code non échec du timer/service ; le transport de cette alerte (courriel, XMPP ou système
nul laisse l'unité `pronote-sync.service` en état `failed` ; la supervision doit de supervision) relève de l'exploitation locale.
donc alerter sur cet état ou sur le code de sortie. Le code `2` permet de
distinguer automatiquement une alerte dégradée d'une panne critique, sans lire
les journaux.
Le `--dry-run` n'écrit ni dans CalDAV/XMPP ni dans l'état local. Il conserve le
même contrat de codes : `0` si la simulation est complète, `2` si elle est
dégradée et `1` si elle est critique.
## Journaux et alertes ## Journaux et alertes
+4 -23
View File
@@ -11,7 +11,6 @@ from pydantic import SecretStr
from pronote_sync.config.env import load_settings from pronote_sync.config.env import load_settings
from pronote_sync.config.settings import Settings from pronote_sync.config.settings import Settings
from pronote_sync.errors import ErrorSeverity, PipelineError
from pronote_sync.pipeline.run import PipelineRunner from pronote_sync.pipeline.run import PipelineRunner
from pronote_sync.utils.logging import setup_logging from pronote_sync.utils.logging import setup_logging
from pronote_sync.utils.redaction import redact_secrets from pronote_sync.utils.redaction import redact_secrets
@@ -20,26 +19,6 @@ logger = logging.getLogger(__name__)
_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") _LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")
# Contrat stable pour systemd et les outils de supervision.
EXIT_SUCCESS = 0
EXIT_CRITICAL = 1
EXIT_DEGRADED = 2
def _pipeline_exit_code(data: object | None, errors: Sequence[PipelineError]) -> int:
"""Convertit le résultat du pipeline en code de sortie supervisable.
:param data: Données normalisées produites, ou ``None`` en cas d'échec critique.
:param errors: Erreurs et avertissements de l'exécution.
:return: ``0`` si complet, ``2`` si dégradé, ``1`` si critique.
:rtype: int
"""
if data is None or any(error.severity == ErrorSeverity.CRITICAL for error in errors):
return EXIT_CRITICAL
if errors:
return EXIT_DEGRADED
return EXIT_SUCCESS
def _parse_arguments(arguments: Sequence[str] | None = None) -> argparse.Namespace: def _parse_arguments(arguments: Sequence[str] | None = None) -> argparse.Namespace:
"""Analyse les options de lancement du programme. """Analyse les options de lancement du programme.
@@ -141,7 +120,7 @@ def main(arguments: Sequence[str] | None = None) -> int:
bruts afin de préserver le diagnostic sans exposer de secret. bruts afin de préserver le diagnostic sans exposer de secret.
:param arguments: Arguments optionnels, principalement utiles aux appels programmatiques. :param arguments: Arguments optionnels, principalement utiles aux appels programmatiques.
:return: Code machine-readable : ``0`` complet, ``2`` dégradé, ``1`` critique. :return: ``0`` en cas de succès, ``1`` sinon (après analyse des arguments).
:rtype: int :rtype: int
:raises SystemExit: Si argparse rejette les arguments (code de sortie 2). :raises SystemExit: Si argparse rejette les arguments (code de sortie 2).
""" """
@@ -168,7 +147,9 @@ def main(arguments: Sequence[str] | None = None) -> int:
secrets = _settings_secrets(settings) secrets = _settings_secrets(settings)
for error in errors: for error in errors:
logger.error("%s", redact_secrets(error.message, extra_secrets=secrets)) logger.error("%s", redact_secrets(error.message, extra_secrets=secrets))
return _pipeline_exit_code(data, errors) if data is None:
return 1
return 0
if __name__ == "__main__": if __name__ == "__main__":
+16 -7
View File
@@ -14,7 +14,7 @@ l'outil et de ne jamais toucher aux événements étrangers du calendrier.
from __future__ import annotations from __future__ import annotations
from datetime import datetime, time from datetime import datetime, timedelta
from typing import cast from typing import cast
from icalendar import Calendar, Component, Event, vDate, vDatetime from icalendar import Calendar, Component, Event, vDate, vDatetime
@@ -33,7 +33,15 @@ MANAGED_VALUE = "v1"
PRODID = "-//pronote-sync//NONSGML v1.0//EN" PRODID = "-//pronote-sync//NONSGML v1.0//EN"
#: Propriétés prises en compte dans la signature sémantique d'un composant. #: Propriétés prises en compte dans la signature sémantique d'un composant.
_SIGNATURE_KEYS: tuple[str, ...] = ("UID", "SUMMARY", "DTSTART", "DTEND", "STATUS", "DESCRIPTION") _SIGNATURE_KEYS: tuple[str, ...] = (
"UID",
"SUMMARY",
"DTSTART",
"DTEND",
"STATUS",
"TRANSP",
"DESCRIPTION",
)
def lesson_to_vevent(lesson: Lesson) -> Event: def lesson_to_vevent(lesson: Lesson) -> Event:
@@ -88,8 +96,9 @@ def lesson_to_vevent(lesson: Lesson) -> Event:
def homework_to_vevent(homework: Homework) -> Event: def homework_to_vevent(homework: Homework) -> Event:
"""Convertit un devoir Pronote en composant VEVENT iCalendar. """Convertit un devoir Pronote en composant VEVENT iCalendar.
Le devoir est représenté comme une tâche (``STATUS:NEEDS-ACTION``) sur la Le devoir est représenté comme un événement toute la journée à la date
journée d'échéance, entre 08:00 et 18:00. d'échéance. Il est transparent pour ne pas bloquer les disponibilités ;
aucun statut de tâche ``VTODO`` n'est ajouté à ce ``VEVENT``.
:param homework: Devoir Pronote à sérialiser. :param homework: Devoir Pronote à sérialiser.
:return: Composant :class:`icalendar.Event` marqué comme géré par l'outil. :return: Composant :class:`icalendar.Event` marqué comme géré par l'outil.
@@ -98,10 +107,10 @@ def homework_to_vevent(homework: Homework) -> Event:
event = Event() event = Event()
event.add("uid", f"homework-{homework.id}") event.add("uid", f"homework-{homework.id}")
event.add("summary", f"Devoir: {homework.subject}") event.add("summary", f"Devoir: {homework.subject}")
event.add("dtstart", vDatetime(datetime.combine(homework.due_on, time(8, 0)))) event.add("dtstart", vDate(homework.due_on))
event.add("dtend", vDatetime(datetime.combine(homework.due_on, time(18, 0)))) event.add("dtend", vDate(homework.due_on + timedelta(days=1)))
event.add("description", homework.text) event.add("description", homework.text)
event.add("status", "NEEDS-ACTION") event.add("transp", "TRANSPARENT")
event.add("categories", ["Pronote", "Devoir"]) event.add("categories", ["Pronote", "Devoir"])
event.add(MANAGED_PROPERTY, MANAGED_VALUE) event.add(MANAGED_PROPERTY, MANAGED_VALUE)
return event return event
+3 -23
View File
@@ -52,10 +52,10 @@ def test_main_runs_composition_root_in_dry_run_with_requested_log_level(
runner.run.assert_called_once_with() runner.run.assert_called_once_with()
def test_main_preserves_configured_dry_run_and_returns_degraded_with_warnings( def test_main_preserves_configured_dry_run_and_returns_success_with_warnings(
mocker: MockerFixture, mocker: MockerFixture,
) -> None: ) -> None:
"""Sans option, la CLI préserve le dry-run configuré et signale l'état dégradé.""" """Sans option, la CLI préserve le dry-run configuré et accepte les avertissements."""
from pronote_sync.cli.main import main from pronote_sync.cli.main import main
settings = Settings(app=AppSettings(dry_run=True, log_level="WARNING")) settings = Settings(app=AppSettings(dry_run=True, log_level="WARNING"))
@@ -72,32 +72,12 @@ def test_main_preserves_configured_dry_run_and_returns_degraded_with_warnings(
exit_code = main([]) exit_code = main([])
assert exit_code == 2 assert exit_code == 0
assert setup_logging.call_args_list == [mocker.call("INFO"), mocker.call("WARNING")] assert setup_logging.call_args_list == [mocker.call("INFO"), mocker.call("WARNING")]
composition_root.assert_called_once_with(settings, dry_run=None) composition_root.assert_called_once_with(settings, dry_run=None)
runner.run.assert_called_once_with() runner.run.assert_called_once_with()
@pytest.mark.parametrize("step", ["caldav_sync", "send"])
def test_main_returns_degraded_code_for_caldav_or_xmpp_failure(
mocker: MockerFixture,
step: str,
) -> None:
"""Les échecs récupérables CalDAV et XMPP sont observables par le code 2."""
from pronote_sync.cli.main import main
settings = Settings()
mocker.patch("pronote_sync.cli.main.load_settings", return_value=settings)
runner = mocker.Mock()
runner.run.return_value = (
mocker.Mock(spec=PronoteData),
[PipelineWarning(f"Échec récupérable de {step}", step=step)],
)
mocker.patch("pronote_sync.cli.main.PipelineRunner.from_settings", return_value=runner)
assert main([]) == 2
def test_main_returns_failure_and_redacts_pipeline_secrets_at_debug_level( def test_main_returns_failure_and_redacts_pipeline_secrets_at_debug_level(
mocker: MockerFixture, mocker: MockerFixture,
capsys: pytest.CaptureFixture[str], capsys: pytest.CaptureFixture[str],
+85 -5
View File
@@ -7,7 +7,7 @@ ajouts/mises à jour/suppressions, et la préservation des événements non gér
from __future__ import annotations from __future__ import annotations
from datetime import datetime from datetime import date, datetime
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
import pytest import pytest
@@ -122,10 +122,16 @@ class FakeCalendar:
event_start = raw_start.dt event_start = raw_start.dt
event_end = raw_end.dt if raw_end is not None else event_start event_end = raw_end.dt if raw_end is not None else event_start
overlaps = True overlaps = True
if start is not None: if isinstance(event_start, datetime):
overlaps = overlaps and event_end > start if start is not None:
if end is not None: overlaps = overlaps and event_end > start
overlaps = overlaps and event_start < end if end is not None:
overlaps = overlaps and event_start < end
else:
if start is not None:
overlaps = overlaps and event_end > start.date()
if end is not None:
overlaps = overlaps and event_start < end.date()
if overlaps: if overlaps:
results.append(FakeCalendarEvent(ical_text, uid=uid, server=self._server)) results.append(FakeCalendarEvent(ical_text, uid=uid, server=self._server))
break break
@@ -686,6 +692,80 @@ class TestCalDAVSynchronize:
assert result.added == 1 assert result.added == 1
assert len(fake_caldav_server.get_events()) == 1 assert len(fake_caldav_server.get_events()) == 1
def test_legacy_homework_vevent_is_migrated_idempotently_and_removed(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Migre un ancien devoir puis vérifie l'idempotence et la suppression."""
fake_caldav_server._events["homework-hw-legacy"] = _create_vevent_text(
uid="homework-hw-legacy",
summary="Devoir: Histoire",
start=datetime(2026, 1, 20, 8, 0),
end=datetime(2026, 1, 20, 18, 0),
status="NEEDS-ACTION",
managed=True,
)
homework = Homework(
id="hw-legacy",
subject="Histoire",
assigned_on=None,
due_on=date(2026, 1, 20),
text="Lire le chapitre 5",
)
def make_data(homeworks: list[Homework]) -> PronoteData:
"""Construit les données de synchronisation du scénario."""
return PronoteData(
lessons=[],
homeworks=homeworks,
school_events=[],
messages=[],
target_date=date(2026, 1, 20),
generated_at=datetime(2026, 1, 14, 0, 0),
)
first = synchronize(
pronote_data=make_data([homework]),
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
assert first.status == CalDAVSyncStatus.SUCCESS
assert first.added == 0
assert first.updated == 1
assert first.removed == 0
migrated = Calendar.from_ical(fake_caldav_server.get_events()["homework-hw-legacy"])
event = migrated.walk("VEVENT")[0]
assert event.get("DTSTART").dt == date(2026, 1, 20)
assert event.get("DTEND").dt == date(2026, 1, 21)
assert event.get("STATUS") is None
assert str(event.get("TRANSP")) == "TRANSPARENT"
second = synchronize(
pronote_data=make_data([homework]),
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
assert second.status == CalDAVSyncStatus.SKIPPED
assert second.added == 0
assert second.updated == 0
assert second.removed == 0
removed = synchronize(
pronote_data=make_data([]),
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
assert removed.status == CalDAVSyncStatus.SUCCESS
assert removed.added == 0
assert removed.updated == 0
assert removed.removed == 1
assert fake_caldav_server.get_events() == {}
def test_school_event_sync( def test_school_event_sync(
self, self,
fake_caldav_server: FakeCalDAVServer, fake_caldav_server: FakeCalDAVServer,
+7 -6
View File
@@ -239,15 +239,16 @@ def test_homework_to_vevent_uid_prefix() -> None:
assert str(event.get("UID")) == "homework-HW-5678" assert str(event.get("UID")) == "homework-HW-5678"
def test_homework_to_vevent_status() -> None: def test_homework_to_vevent_is_transparent_without_task_status() -> None:
"""Vérifie qu'un devoir a STATUS=NEEDS-ACTION. """Vérifie qu'un devoir VEVENT est transparent et sans statut VTODO.
:return: None :return: None
""" """
homework = _make_homework() homework = _make_homework()
event = homework_to_vevent(homework) event = homework_to_vevent(homework)
assert str(event.get("STATUS")) == "NEEDS-ACTION" assert event.get("STATUS") is None
assert str(event.get("TRANSP")) == "TRANSPARENT"
def test_homework_to_vevent_categories() -> None: def test_homework_to_vevent_categories() -> None:
@@ -265,15 +266,15 @@ def test_homework_to_vevent_categories() -> None:
def test_homework_to_vevent_dtstart_dtend() -> None: def test_homework_to_vevent_dtstart_dtend() -> None:
"""Vérifie que DTSTART et DTEND couvrent la journée d'échéance (08:00-18:00). """Vérifie que DTSTART et DTEND encadrent la journée d'échéance.
:return: None :return: None
""" """
homework = _make_homework(due_on=date(2026, 1, 20)) homework = _make_homework(due_on=date(2026, 1, 20))
event = homework_to_vevent(homework) event = homework_to_vevent(homework)
assert event.get("DTSTART").dt == datetime(2026, 1, 20, 8, 0) assert event.get("DTSTART").dt == date(2026, 1, 20)
assert event.get("DTEND").dt == datetime(2026, 1, 20, 18, 0) assert event.get("DTEND").dt == date(2026, 1, 21)
def test_homework_to_vevent_summary() -> None: def test_homework_to_vevent_summary() -> None: