Compare commits
1 Commits
feature/m1
...
m12-cli-en
| Author | SHA1 | Date | |
|---|---|---|---|
|
fd9b604849
|
8
TODO.md
8
TODO.md
@@ -241,10 +241,10 @@ Composer et orchestrer toutes les étapes avec gestion d'erreurs dégradée et m
|
|||||||
|
|
||||||
Exposer le lancement du pipeline via une interface en ligne de commande.
|
Exposer le lancement du pipeline via une interface en ligne de commande.
|
||||||
|
|
||||||
- [ ] Créer `cli/main.py` : `main()` (point d'entrée `pronote-sync`), args `--dry-run`, `--log-level`.
|
- [x] Créer `cli/main.py` : `main()` (point d'entrée `pronote-sync`), args `--dry-run`, `--log-level`.
|
||||||
- [ ] Initialiser les logs (`setup_logging`) et charger `settings` au démarrage.
|
- [x] Initialiser les logs (`setup_logging`) et charger `settings` au démarrage.
|
||||||
- [ ] Construire la composition root et lancer `PipelineRunner.run()`.
|
- [x] Construire la composition root et lancer `PipelineRunner.run()`.
|
||||||
- [ ] Gérer le code de retour et l'affichage des erreurs (redactées).
|
- [x] Gérer le code de retour et l'affichage des erreurs (redactées).
|
||||||
|
|
||||||
### Critères d'acceptation
|
### Critères d'acceptation
|
||||||
- `pronote-sync --dry-run --log-level DEBUG` s'exécute sans effet de bord.
|
- `pronote-sync --dry-run --log-level DEBUG` s'exécute sans effet de bord.
|
||||||
|
|||||||
@@ -0,0 +1 @@
|
|||||||
|
"""Interface en ligne de commande du pipeline ``pronote-sync``."""
|
||||||
|
|||||||
153
pronote_sync/cli/main.py
Normal file
153
pronote_sync/cli/main.py
Normal file
@@ -0,0 +1,153 @@
|
|||||||
|
"""Point d'entrée en ligne de commande du pipeline Pronote → CalDAV → XMPP."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
import logging
|
||||||
|
import traceback
|
||||||
|
from collections.abc import Sequence
|
||||||
|
|
||||||
|
from pydantic import SecretStr
|
||||||
|
|
||||||
|
from pronote_sync.config.env import load_settings
|
||||||
|
from pronote_sync.config.settings import Settings
|
||||||
|
from pronote_sync.pipeline.run import PipelineRunner
|
||||||
|
from pronote_sync.utils.logging import setup_logging
|
||||||
|
from pronote_sync.utils.redaction import redact_secrets
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")
|
||||||
|
|
||||||
|
|
||||||
|
def _parse_arguments(arguments: Sequence[str] | None = None) -> argparse.Namespace:
|
||||||
|
"""Analyse les options de lancement du programme.
|
||||||
|
|
||||||
|
:param arguments: Arguments à analyser, ou ``None`` pour ceux du processus.
|
||||||
|
:return: Options de ligne de commande validées.
|
||||||
|
:rtype: argparse.Namespace
|
||||||
|
"""
|
||||||
|
parser = argparse.ArgumentParser(description="Synchronise Pronote vers CalDAV et XMPP.")
|
||||||
|
parser.add_argument(
|
||||||
|
"--dry-run",
|
||||||
|
action="store_true",
|
||||||
|
default=None,
|
||||||
|
help="Simule la synchronisation sans écrire vers CalDAV ni XMPP.",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--log-level",
|
||||||
|
choices=_LOG_LEVELS,
|
||||||
|
type=str.upper,
|
||||||
|
help="Niveau de verbosité des journaux.",
|
||||||
|
)
|
||||||
|
return parser.parse_args(arguments)
|
||||||
|
|
||||||
|
|
||||||
|
def _settings_secrets(settings: Settings) -> tuple[SecretStr | str, ...]:
|
||||||
|
"""Retourne les valeurs sensibles connues pour la rédaction des messages.
|
||||||
|
|
||||||
|
Centraliser ces valeurs garantit que les diagnostics CLI ne divulguent pas
|
||||||
|
les secrets configurés, y compris lorsque le niveau ``DEBUG`` est demandé.
|
||||||
|
|
||||||
|
:param settings: Configuration validée de l'application.
|
||||||
|
:return: Secrets connus à transmettre au mécanisme de rédaction.
|
||||||
|
:rtype: tuple[SecretStr | str, ...]
|
||||||
|
"""
|
||||||
|
candidates = (
|
||||||
|
*settings.redaction_secrets(),
|
||||||
|
settings.pronote.username,
|
||||||
|
settings.caldav.username,
|
||||||
|
settings.xmpp.jid,
|
||||||
|
settings.xmpp.to,
|
||||||
|
)
|
||||||
|
return tuple(dict.fromkeys(secret for secret in candidates if secret is not None))
|
||||||
|
|
||||||
|
|
||||||
|
def _safe_traceback(
|
||||||
|
exception: BaseException, *, extra_secrets: Sequence[SecretStr | str] = ()
|
||||||
|
) -> str:
|
||||||
|
"""Construit une pile complète sans inclure les messages d'exception bruts.
|
||||||
|
|
||||||
|
Les noms de fichiers, lignes et fonctions conservent la valeur de diagnostic
|
||||||
|
de la pile. Les messages et les chaînes de causes sont volontairement
|
||||||
|
remplacés, car ils peuvent provenir d'une bibliothèque externe.
|
||||||
|
|
||||||
|
:param exception: Exception à représenter sans divulguer son contenu.
|
||||||
|
:param extra_secrets: Valeurs sensibles configurées à rédiger dans les cadres.
|
||||||
|
:return: Représentation de la pile et de ses causes, expurgée.
|
||||||
|
:rtype: str
|
||||||
|
"""
|
||||||
|
lines = ["Traceback (most recent call last):"]
|
||||||
|
current: BaseException | None = exception
|
||||||
|
seen: set[int] = set()
|
||||||
|
while current is not None and id(current) not in seen:
|
||||||
|
seen.add(id(current))
|
||||||
|
for frame in traceback.extract_tb(current.__traceback__):
|
||||||
|
lines.append(f' File "{frame.filename}", line {frame.lineno}, in {frame.name}')
|
||||||
|
lines.append(f"{type(current).__name__}: erreur expurgée")
|
||||||
|
next_exception = current.__cause__ or current.__context__
|
||||||
|
if next_exception is not None and id(next_exception) not in seen:
|
||||||
|
lines.append("La cause ou le contexte précédent est le suivant :")
|
||||||
|
current = next_exception
|
||||||
|
return redact_secrets("\n".join(lines), extra_secrets=extra_secrets)
|
||||||
|
|
||||||
|
|
||||||
|
def _log_failure(
|
||||||
|
message: str,
|
||||||
|
exception: BaseException,
|
||||||
|
*,
|
||||||
|
extra_secrets: Sequence[SecretStr | str] = (),
|
||||||
|
) -> None:
|
||||||
|
"""Journalise une erreur et sa pile expurgée uniquement en niveau DEBUG.
|
||||||
|
|
||||||
|
:param message: Message public déjà sûr à afficher hors DEBUG.
|
||||||
|
:param exception: Exception dont la pile doit être présentée de façon sûre.
|
||||||
|
:param extra_secrets: Valeurs sensibles configurées à rédiger.
|
||||||
|
:rtype: None
|
||||||
|
"""
|
||||||
|
logger.error("%s", redact_secrets(message, extra_secrets=extra_secrets))
|
||||||
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
|
logger.debug("%s", _safe_traceback(exception, extra_secrets=extra_secrets))
|
||||||
|
|
||||||
|
|
||||||
|
def main(arguments: Sequence[str] | None = None) -> int:
|
||||||
|
"""Lance le pipeline configuré et retourne son code de sortie.
|
||||||
|
|
||||||
|
En niveau ``DEBUG``, les piles sont affichées sans leurs messages externes
|
||||||
|
bruts afin de préserver le diagnostic sans exposer de secret.
|
||||||
|
|
||||||
|
:param arguments: Arguments optionnels, principalement utiles aux appels programmatiques.
|
||||||
|
:return: ``0`` en cas de succès, ``1`` sinon (après analyse des arguments).
|
||||||
|
:rtype: int
|
||||||
|
:raises SystemExit: Si argparse rejette les arguments (code de sortie 2).
|
||||||
|
"""
|
||||||
|
parsed_arguments = _parse_arguments(arguments)
|
||||||
|
setup_logging(parsed_arguments.log_level or "INFO")
|
||||||
|
try:
|
||||||
|
settings = load_settings()
|
||||||
|
except Exception as exception:
|
||||||
|
_log_failure("Configuration invalide ou indisponible.", exception)
|
||||||
|
return 1
|
||||||
|
|
||||||
|
setup_logging(parsed_arguments.log_level or settings.app.log_level)
|
||||||
|
try:
|
||||||
|
runner = PipelineRunner.from_settings(settings, dry_run=parsed_arguments.dry_run)
|
||||||
|
data, errors = runner.run()
|
||||||
|
except Exception as exception:
|
||||||
|
_log_failure(
|
||||||
|
"Échec inattendu du pipeline.",
|
||||||
|
exception,
|
||||||
|
extra_secrets=_settings_secrets(settings),
|
||||||
|
)
|
||||||
|
return 1
|
||||||
|
|
||||||
|
secrets = _settings_secrets(settings)
|
||||||
|
for error in errors:
|
||||||
|
logger.error("%s", redact_secrets(error.message, extra_secrets=secrets))
|
||||||
|
if data is None:
|
||||||
|
return 1
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
raise SystemExit(main())
|
||||||
1
tests/e2e/__init__.py
Normal file
1
tests/e2e/__init__.py
Normal file
@@ -0,0 +1 @@
|
|||||||
|
"""Tests end-to-end de l'interface en ligne de commande."""
|
||||||
189
tests/e2e/test_cli.py
Normal file
189
tests/e2e/test_cli.py
Normal file
@@ -0,0 +1,189 @@
|
|||||||
|
"""Tests de l'interface en ligne de commande ``pronote-sync``."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from pydantic import SecretStr
|
||||||
|
from pytest_mock import MockerFixture
|
||||||
|
|
||||||
|
from pronote_sync.config.settings import AISettings, AppSettings, PronoteSettings, Settings
|
||||||
|
from pronote_sync.errors import PipelineCriticalError, PipelineWarning
|
||||||
|
from pronote_sync.models.pronote import PronoteData
|
||||||
|
|
||||||
|
|
||||||
|
def test_main_runs_composition_root_in_dry_run_with_requested_log_level(
|
||||||
|
mocker: MockerFixture,
|
||||||
|
) -> None:
|
||||||
|
"""La CLI propage les options au logger et au runner injecté."""
|
||||||
|
from pronote_sync.cli.main import main
|
||||||
|
|
||||||
|
settings = Settings(app=AppSettings(log_level="WARNING"))
|
||||||
|
load_settings = mocker.patch("pronote_sync.cli.main.load_settings", return_value=settings)
|
||||||
|
setup_logging = mocker.patch("pronote_sync.cli.main.setup_logging")
|
||||||
|
runner = mocker.Mock()
|
||||||
|
runner.run.return_value = (mocker.Mock(spec=PronoteData), [])
|
||||||
|
composition_root = mocker.patch(
|
||||||
|
"pronote_sync.cli.main.PipelineRunner.from_settings", return_value=runner
|
||||||
|
)
|
||||||
|
|
||||||
|
exit_code = main(["--dry-run", "--log-level", "DEBUG"])
|
||||||
|
|
||||||
|
assert exit_code == 0
|
||||||
|
load_settings.assert_called_once_with()
|
||||||
|
assert setup_logging.call_args_list == [mocker.call("DEBUG"), mocker.call("DEBUG")]
|
||||||
|
composition_root.assert_called_once_with(settings, dry_run=True)
|
||||||
|
runner.run.assert_called_once_with()
|
||||||
|
|
||||||
|
|
||||||
|
def test_main_preserves_configured_dry_run_and_returns_success_with_warnings(
|
||||||
|
mocker: MockerFixture,
|
||||||
|
) -> None:
|
||||||
|
"""Sans option, la CLI préserve le dry-run configuré et accepte les avertissements."""
|
||||||
|
from pronote_sync.cli.main import main
|
||||||
|
|
||||||
|
settings = Settings(app=AppSettings(dry_run=True, log_level="WARNING"))
|
||||||
|
mocker.patch("pronote_sync.cli.main.load_settings", return_value=settings)
|
||||||
|
setup_logging = mocker.patch("pronote_sync.cli.main.setup_logging")
|
||||||
|
runner = mocker.Mock()
|
||||||
|
runner.run.return_value = (
|
||||||
|
mocker.Mock(spec=PronoteData),
|
||||||
|
[PipelineWarning("Avertissement non bloquant")],
|
||||||
|
)
|
||||||
|
composition_root = mocker.patch(
|
||||||
|
"pronote_sync.cli.main.PipelineRunner.from_settings", return_value=runner
|
||||||
|
)
|
||||||
|
|
||||||
|
exit_code = main([])
|
||||||
|
|
||||||
|
assert exit_code == 0
|
||||||
|
assert setup_logging.call_args_list == [mocker.call("INFO"), mocker.call("WARNING")]
|
||||||
|
composition_root.assert_called_once_with(settings, dry_run=None)
|
||||||
|
runner.run.assert_called_once_with()
|
||||||
|
|
||||||
|
|
||||||
|
def test_main_returns_failure_and_redacts_pipeline_secrets_at_debug_level(
|
||||||
|
mocker: MockerFixture,
|
||||||
|
capsys: pytest.CaptureFixture[str],
|
||||||
|
) -> None:
|
||||||
|
"""Les diagnostics de pipeline restent expurgés, même au niveau DEBUG."""
|
||||||
|
from pronote_sync.cli.main import main
|
||||||
|
|
||||||
|
secret = "M12_PIPELINE_SECRET" # pragma: allowlist secret
|
||||||
|
settings = Settings(ai=AISettings(api_key=SecretStr(secret)))
|
||||||
|
mocker.patch("pronote_sync.cli.main.load_settings", return_value=settings)
|
||||||
|
runner = mocker.Mock()
|
||||||
|
runner.run.return_value = (
|
||||||
|
None,
|
||||||
|
[PipelineCriticalError(f"Échec distant avec le secret {secret}")],
|
||||||
|
)
|
||||||
|
mocker.patch("pronote_sync.cli.main.PipelineRunner.from_settings", return_value=runner)
|
||||||
|
|
||||||
|
exit_code = main(["--log-level", "DEBUG"])
|
||||||
|
|
||||||
|
output = capsys.readouterr().out
|
||||||
|
assert exit_code == 1
|
||||||
|
assert secret not in output
|
||||||
|
assert "REDACTED" in output
|
||||||
|
assert "Traceback" not in output
|
||||||
|
|
||||||
|
|
||||||
|
def test_main_displays_a_redacted_configuration_traceback_at_debug_level(
|
||||||
|
mocker: MockerFixture,
|
||||||
|
capsys: pytest.CaptureFixture[str],
|
||||||
|
) -> None:
|
||||||
|
"""Une erreur de configuration DEBUG conserve son traceback sans son secret."""
|
||||||
|
from pronote_sync.cli.main import main
|
||||||
|
|
||||||
|
secret = "M12_CONFIGURATION_SECRET" # pragma: allowlist secret
|
||||||
|
mocker.patch(
|
||||||
|
"pronote_sync.cli.main.load_settings",
|
||||||
|
side_effect=ValueError(f"configuration invalide: {secret}"),
|
||||||
|
)
|
||||||
|
|
||||||
|
exit_code = main(["--log-level", "DEBUG"])
|
||||||
|
|
||||||
|
output = capsys.readouterr().out
|
||||||
|
assert exit_code == 1
|
||||||
|
assert secret not in output
|
||||||
|
assert "Configuration invalide ou indisponible." in output
|
||||||
|
assert "Traceback" in output
|
||||||
|
|
||||||
|
|
||||||
|
def test_main_does_not_disclose_a_configured_pronote_username(
|
||||||
|
mocker: MockerFixture,
|
||||||
|
capsys: pytest.CaptureFixture[str],
|
||||||
|
) -> None:
|
||||||
|
"""Les erreurs critiques ne divulguent pas un identifiant Pronote configuré."""
|
||||||
|
from pronote_sync.cli.main import main
|
||||||
|
|
||||||
|
username = "m12-parent-identifier"
|
||||||
|
settings = Settings(pronote=PronoteSettings(username=username))
|
||||||
|
mocker.patch("pronote_sync.cli.main.load_settings", return_value=settings)
|
||||||
|
runner = mocker.Mock()
|
||||||
|
runner.run.return_value = (
|
||||||
|
None,
|
||||||
|
[PipelineCriticalError(f"Échec distant pour l'identifiant {username}")],
|
||||||
|
)
|
||||||
|
mocker.patch("pronote_sync.cli.main.PipelineRunner.from_settings", return_value=runner)
|
||||||
|
|
||||||
|
exit_code = main([])
|
||||||
|
|
||||||
|
output = capsys.readouterr().out
|
||||||
|
assert exit_code == 1
|
||||||
|
assert username not in output
|
||||||
|
assert "REDACTED" in output
|
||||||
|
|
||||||
|
|
||||||
|
def test_main_rejects_an_unknown_log_level() -> None:
|
||||||
|
"""La CLI rejette les niveaux de journalisation hors contrat."""
|
||||||
|
from pronote_sync.cli.main import main
|
||||||
|
|
||||||
|
with pytest.raises(SystemExit) as error:
|
||||||
|
main(["--log-level", "VERBOSE"])
|
||||||
|
|
||||||
|
assert error.value.code == 2
|
||||||
|
|
||||||
|
|
||||||
|
def test_main_logs_redacted_traceback_when_pipeline_raises_unexpectedly(
|
||||||
|
mocker: MockerFixture,
|
||||||
|
capsys: pytest.CaptureFixture[str],
|
||||||
|
) -> None:
|
||||||
|
"""Une exception inattendue du pipeline produit un traceback expurgé en DEBUG."""
|
||||||
|
from pronote_sync.cli.main import main
|
||||||
|
|
||||||
|
secret = "M12_UNEXPECTED_SECRET" # pragma: allowlist secret
|
||||||
|
settings = Settings(ai=AISettings(api_key=SecretStr(secret)))
|
||||||
|
mocker.patch("pronote_sync.cli.main.load_settings", return_value=settings)
|
||||||
|
mocker.patch(
|
||||||
|
"pronote_sync.cli.main.PipelineRunner.from_settings",
|
||||||
|
side_effect=RuntimeError(f"Erreur interne avec {secret}"),
|
||||||
|
)
|
||||||
|
|
||||||
|
exit_code = main(["--log-level", "DEBUG"])
|
||||||
|
|
||||||
|
output = capsys.readouterr().out
|
||||||
|
assert exit_code == 1
|
||||||
|
assert secret not in output
|
||||||
|
assert "Traceback" in output
|
||||||
|
assert "erreur expurgée" in output
|
||||||
|
|
||||||
|
|
||||||
|
def test_main_does_not_show_traceback_at_info_level(
|
||||||
|
mocker: MockerFixture,
|
||||||
|
capsys: pytest.CaptureFixture[str],
|
||||||
|
) -> None:
|
||||||
|
"""En niveau INFO, aucune pile n'est affichée pour une erreur inattendue."""
|
||||||
|
from pronote_sync.cli.main import main
|
||||||
|
|
||||||
|
mocker.patch("pronote_sync.cli.main.load_settings", return_value=Settings())
|
||||||
|
mocker.patch(
|
||||||
|
"pronote_sync.cli.main.PipelineRunner.from_settings",
|
||||||
|
side_effect=RuntimeError("Erreur interne"),
|
||||||
|
)
|
||||||
|
|
||||||
|
exit_code = main([])
|
||||||
|
|
||||||
|
output = capsys.readouterr().out
|
||||||
|
assert exit_code == 1
|
||||||
|
assert "Traceback" not in output
|
||||||
|
assert "Échec inattendu du pipeline." in output
|
||||||
Reference in New Issue
Block a user