feat(M12): CLI entry point with dry-run, log-level, redacted error display
Implement the CLI entry point for pronote-sync: cli/main.py: - main() entry point with --dry-run (tri-state: None defers to settings, True overrides) and --log-level (choices: DEBUG/INFO/WARNING/ERROR/CRITICAL) - setup_logging called before settings load (to capture config errors), then reconfigured with settings.app.log_level - PipelineRunner.from_settings() as composition root, runner.run() - Return codes: 0 success, 1 failure, 2 argparse rejection - _safe_traceback: strips exception messages, replaces with "erreur expurgée", walks __cause__/__context__ with cycle protection - _settings_secrets: collects redaction_secrets() + usernames + JID/recipient - All error messages redacted via redact_secrets() with configured secrets - DEBUG-level traceback only shown when DEBUG is enabled cli/__init__.py: - Module docstring added (French, Sphinx/reST) tests/e2e/test_cli.py (8 tests): - Dry-run and log-level propagation to composition root - Configured dry-run preserved (tri-state None) - Success with warnings returns 0 - Pipeline error redaction at DEBUG (sentinel secret) - Configuration failure redacted traceback at DEBUG - Pronote username non-disclosure - Unexpected pipeline exception: redacted traceback at DEBUG, no traceback at INFO - Argparse rejection of unknown log level (exit code 2) Coverage: cli/ 94.74%, 627 total tests pass. Co-authored-by: opencode/coder <coder@agents.invalid> Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
This commit is contained in:
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())
|
||||
Reference in New Issue
Block a user