Compare commits

..
Author SHA1 Message Date
Codex 5bd97402bf fix(cli): exposer les executions degradees 2026-09-12 15:37:11 +02:00
6 changed files with 75 additions and 119 deletions
+11 -4
View File
@@ -105,10 +105,17 @@ sudo systemctl start pronote-sync.service
sudo systemctl status pronote-sync.service sudo systemctl status pronote-sync.service
``` ```
Une exécution en échec laisse l'unité `pronote-sync.service` en état `failed`. La CLI expose un contrat de sortie stable : `0` signifie une exécution complète,
La supervision de l'hôte doit donc déclencher une alerte sur cet état ou sur un `2` une exécution dégradée (les données Pronote sont disponibles mais une étape
échec du timer/service ; le transport de cette alerte (courriel, XMPP ou système optionnelle, CalDAV ou XMPP a échoué), et `1` un échec critique. Tout code non
de supervision) relève de l'exploitation locale. nul laisse l'unité `pronote-sync.service` en état `failed` ; la supervision doit
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
+23 -4
View File
@@ -11,6 +11,7 @@ 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
@@ -19,6 +20,26 @@ 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.
@@ -120,7 +141,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: ``0`` en cas de succès, ``1`` sinon (après analyse des arguments). :return: Code machine-readable : ``0`` complet, ``2`` dégradé, ``1`` critique.
: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).
""" """
@@ -147,9 +168,7 @@ 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))
if data is None: return _pipeline_exit_code(data, errors)
return 1
return 0
if __name__ == "__main__": if __name__ == "__main__":
+7 -16
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, timedelta from datetime import datetime, time
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,15 +33,7 @@ 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, ...] = ( _SIGNATURE_KEYS: tuple[str, ...] = ("UID", "SUMMARY", "DTSTART", "DTEND", "STATUS", "DESCRIPTION")
"UID",
"SUMMARY",
"DTSTART",
"DTEND",
"STATUS",
"TRANSP",
"DESCRIPTION",
)
def lesson_to_vevent(lesson: Lesson) -> Event: def lesson_to_vevent(lesson: Lesson) -> Event:
@@ -96,9 +88,8 @@ 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 un événement toute la journée à la date Le devoir est représenté comme une tâche (``STATUS:NEEDS-ACTION``) sur la
d'échéance. Il est transparent pour ne pas bloquer les disponibilités ; journée d'échéance, entre 08:00 et 18:00.
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.
@@ -107,10 +98,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", vDate(homework.due_on)) event.add("dtstart", vDatetime(datetime.combine(homework.due_on, time(8, 0))))
event.add("dtend", vDate(homework.due_on + timedelta(days=1))) event.add("dtend", vDatetime(datetime.combine(homework.due_on, time(18, 0))))
event.add("description", homework.text) event.add("description", homework.text)
event.add("transp", "TRANSPARENT") event.add("status", "NEEDS-ACTION")
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
+23 -3
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_success_with_warnings( def test_main_preserves_configured_dry_run_and_returns_degraded_with_warnings(
mocker: MockerFixture, mocker: MockerFixture,
) -> None: ) -> None:
"""Sans option, la CLI préserve le dry-run configuré et accepte les avertissements.""" """Sans option, la CLI préserve le dry-run configuré et signale l'état dégradé."""
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,12 +72,32 @@ def test_main_preserves_configured_dry_run_and_returns_success_with_warnings(
exit_code = main([]) exit_code = main([])
assert exit_code == 0 assert exit_code == 2
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],
+5 -85
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 date, datetime from datetime import datetime
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
import pytest import pytest
@@ -122,16 +122,10 @@ 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 isinstance(event_start, datetime): if start is not None:
if start is not None: overlaps = overlaps and event_end > start
overlaps = overlaps and event_end > start if end is not None:
if end is not None: overlaps = overlaps and event_start < end
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
@@ -692,80 +686,6 @@ 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,
+6 -7
View File
@@ -239,16 +239,15 @@ 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_is_transparent_without_task_status() -> None: def test_homework_to_vevent_status() -> None:
"""Vérifie qu'un devoir VEVENT est transparent et sans statut VTODO. """Vérifie qu'un devoir a STATUS=NEEDS-ACTION.
:return: None :return: None
""" """
homework = _make_homework() homework = _make_homework()
event = homework_to_vevent(homework) event = homework_to_vevent(homework)
assert event.get("STATUS") is None assert str(event.get("STATUS")) == "NEEDS-ACTION"
assert str(event.get("TRANSP")) == "TRANSPARENT"
def test_homework_to_vevent_categories() -> None: def test_homework_to_vevent_categories() -> None:
@@ -266,15 +265,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 encadrent la journée d'échéance. """Vérifie que DTSTART et DTEND couvrent la journée d'échéance (08:00-18:00).
: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 == date(2026, 1, 20) assert event.get("DTSTART").dt == datetime(2026, 1, 20, 8, 0)
assert event.get("DTEND").dt == date(2026, 1, 21) assert event.get("DTEND").dt == datetime(2026, 1, 20, 18, 0)
def test_homework_to_vevent_summary() -> None: def test_homework_to_vevent_summary() -> None: