fix(cli) : rendre les exécutions dégradées observables #40

Merged
OpenCode merged 1 commits from fix/issue-13-observable-status into main 2026-09-12 23:53:58 +02:00
3 changed files with 57 additions and 11 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__":
+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],