Compare commits

...

18 Commits

Author SHA1 Message Date
de647a6ff8 docs: ajouter la politique de versionnage et releases dans AGENTS.md
Ajoute la section 13 « Versionnage et releases » qui documente :
- Politique semver en phase 0.x (patch / minor / 1.0.0)
- Règle absolue : pas de tag sans validation en environnement réel
- Procédure de release en 7 étapes
- Cohérence tag Git / pyproject.toml / CHANGELOG.md
2026-09-08 21:04:08 +02:00
82b9877aad fix: PRONOTE_ENT optionnel pour les connexions pronotepy directes
PRONOTE_ENT était incorrectement traité comme obligatoire pour pronotepy. Rend ENT optionnel pour les connexions directes, conformément à la spécification. Corrige le mode pronotepy explicite et le mode auto sans iCal. 10 tests de régression ajoutés.

Co-authored-by: Antoine Van Elstraete <antoine@van-elstraete.net>
Co-committed-by: Antoine Van Elstraete <antoine@van-elstraete.net>
2026-09-08 20:54:03 +02:00
c851f67172 docs: fichier de vacances scolaires Bordeaux (zone A, 2026-2027)
Crée data/school_holidays.json avec les dates officielles de l'académie de Bordeaux (zone A) pour 2026-2027. Corrige l'erreur au démarrage quand SCHOOL_HOLIDAYS_PATH pointe vers un fichier inexistant. Documentation wiki : page GuideAgendaVacancesScolaires créée et liée dans le sidebar.

Co-authored-by: Antoine Van Elstraete <antoine@van-elstraete.net>
Co-committed-by: Antoine Van Elstraete <antoine@van-elstraete.net>
2026-09-08 20:30:47 +02:00
4ac5be4c8d fix: corrige l'attribution de pronote-digest vers Yoan Bernabeu
L'attribution pointait à tort vers Antoine Coulon et le dépôt
antoine-coulon/pronote-digest. Le projet pronote-digest a été créé par
Yoan Bernabeu (https://yoanbernabeu.github.io/pronote-digest/, dépôt
https://github.com/yoanbernabeu/pronote-digest).

Co-authored-by: opencode/orchestrator <opencode-orchestrator@agents.invalid>
2026-09-08 18:57:55 +02:00
fdd3310462 fix: corrige l'attribution de pronote-digest vers Yoan Bernabeu
L'attribution pointait à tort vers Antoine Coulon et le dépôt
antoine-coulon/pronote-digest. Le projet pronote-digest a été créé par
Yoan Bernabeu (https://yoanbernabeu.github.io/pronote-digest/, dépôt
https://github.com/yoanbernabeu/pronote-digest).

Co-authored-by: opencode/orchestrator <opencode-orchestrator@agents.invalid>
2026-09-08 18:56:44 +02:00
0be02a660a docs: add M15 (Documentation) to CHANGELOG 0.1.0 entry 2026-09-08 18:33:40 +02:00
d85733116a chore: ignore HANDOFF.md and clean up temporary work files
Add HANDOFF.md to .gitignore for session handoff notes.
Delete 12 temporary files (FIXME_M1-M10, FEAT_M9, TEST_REPORT).
2026-09-08 17:55:55 +02:00
2deeb83c76 feat(M15): README, README.LLM.md, LICENSE, CHANGELOG, and final cleanup
Complete the M15 documentation and final review milestone:

README.md (French, brief):
- Project description, quick start, usage, deployment link, acknowledgments
- Acknowledgments: pronotepy, icalendar, caldav, slixmpp, pydantic, openai,
  feedparser, beautifulsoup4, and inspiration from pronote-digest (Antoine Coulon)
- Author: Antoine Van Elstraete

README.LLM.md (English, AI agent guide):
- Prerequisites, installation on LXC/VPS (Debian/CentOS)
- Non-secret configuration preparation (all env vars listed)
- Secret variables clearly identified as operator-only
- Pre-deployment checks (check_secrets.py, pip check, dry-run)
- systemd/timer/logrotate installation steps
- Notes for AI agent (do not commit secrets, do not modify existing files)

LICENSE:
- Standard MIT License, copyright Antoine Van Elstraete (2026)

CHANGELOG.md:
- Initial changelog following Keep a Changelog format
- Covers M1 through M14 with milestone summaries
- Gitea Actions noted as planned/optional, not delivered

.gitignore:
- Added FEAT_* pattern for temporary work files

TODO.md:
- M15 items 1-4 checked (README, architecture docs, CHANGELOG+LICENSE, final review)
- M15 item 5 (Gitea Actions) left unchecked (optional, not done)

Co-authored-by: opencode/tech-writer <tech-writer@agents.invalid>
2026-09-08 17:52:45 +02:00
b518508632 chore: replace GitHub Actions CI/CD with Gitea Actions (LXC/VPS)
Replace GitHub Actions references with Gitea Actions for deployment
on LXC/VPS (Debian/CentOS):

TODO.md:
- M15 optional item: GitHub Actions CI/CD -> Gitea Actions (LXC/VPS)
- Acceptance criterion: "La CI exécute..." -> "Gitea Actions exécute..."

GUIDE_DEV_PYTHON.md:
- §Prochaines étapes: "Configurer CI/CD" with GitHub Actions -> Gitea
  Actions for LXC/VPS (Debian/CentOS)

Other github.com references (library URLs, pre-commit repos, project
URLs) remain unchanged — they are legitimate technical references.
2026-09-08 17:33:17 +02:00
b474f02e90 fix(M14): harden secret scanner — prefixed vars, index blobs, extensionless files
Correct five findings from independent review and security audit of the
M14 deployment secret scanner:

scripts/check_secrets.py:
- Regex: \b[a-z0-9_]* prefix before sensitive keywords to detect
  PRONOTE_PASSWORD, CALDAV_PASSWORD, AI_API_KEY and similar prefixed
  variable names (was: \b which doesn't match before underscore)
- Regex: minimum secret value length reduced from {8,} to {3,} for
  literal, unquoted, and URL parameter patterns
- Regex: unquoted pattern {3,} -> {2,} for 3-char total minimum
- --staged: reads Git index blobs via `git show :<path>` instead of
  working-tree files (ContentProvider type alias, _staged_content_provider)
- Extensionless deployment files: _EXTRA_NAMES allowlist for pronote_sync
- ContentProvider type alias documented with #: Sphinx comment

tests/unit/test_check_secrets.py (4 new tests, 9 total):
- test_main_detects_prefixed_secret_assignment: PRONOTE_PASSWORD detected
- test_main_detects_short_secret_assignment: 6-char secret detected
- test_staged_mode_reads_index_content_not_working_tree: working-tree
  content set to non-matching value to distinguish index from worktree
- test_main_scans_extensionless_deployment_file: pronote_sync scanned

TODO.md: all 5 M14 checklist items checked

Validation: 636 tests, coverage 95.67%, ruff/mypy/bandit/pre-commit green.

Co-authored-by: opencode/coder <coder@agents.invalid>
2026-09-08 17:20:08 +02:00
e6e4b10047 Merge branch 'main' into feature/m14-deployment 2026-09-08 16:43:15 +02:00
d60357a017 feat(M14): add deployment artifacts and secret check
Co-authored-by: Codex/gpt-5.6-terra <codex-gpt-5.6-terra@agents.invalid>
2026-09-08 16:42:59 +02:00
000416f24e feat(M13): complete test fixtures, shared conftest, and coverage >= 90%
Finalize M13 test and coverage milestone:

tests/fixtures/pronote-6e.ics (new):
- Anonymized iCal fixture for Classe de 6e (3 VEVENTs: SVT lesson with
  homework block, Histoire-Géo modified lesson, all-day school outing)
- Same structure as pronote-4e.ics, no secrets or real data

tests/conftest.py:
- Added sample_message fixture (Message with MessageType.INFORMATION)
- Integrated sample_message into pronote_data fixture (messages=[sample_message])
- Sphinx/reST docstring with :return: and :rtype:

.gitignore:
- Fixed typo: .worktress/ -> .worktrees/ (line 56)

pyproject.toml:
- Added ".worktrees" to ruff extend-exclude to prevent ruff format --check .
  from scanning worktree files

TODO.md:
- Checked M13 items: tests/fixtures/ and tests/conftest.py

Validation: 627 tests pass, coverage 95.67% (threshold 90%), ruff/mypy/
bandit/pre-commit all green.

Co-authored-by: opencode/coder <coder@agents.invalid>
2026-09-08 16:19:27 +02:00
fd9b604849 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>
2026-09-08 15:57:28 +02:00
1019b22808 docs(M11): align GUIDE sections 4.2.1 and 11.3 with FIXME_M11 corrections
GUIDE_DEV_PYTHON.md:
- §11.3: add except PipelineCriticalError: raise before each non-critical
  except in the illustrative PipelineRunner.run() code
- §11.3: replace redact_exception(exc) with self._redact(exc) in all except
  blocks, add explanatory paragraph about _redaction_secrets and _redact()
- §11.3: fix Google-style Returns: to Sphinx/reST :return: and :rtype:
- §11.3: fix malformed Markdown code fence (get_errors/get_warnings orphaned)
- §4.2.1: fix redact_exception() example to pass extra_secrets to
  redact_secrets() in the return statement

TODO.md M11:
- Add and check criterion: PipelineCriticalError from non-blocking step
  stops the pipeline

.secrets.baseline:
- Line numbers updated for documentation shifts

Co-authored-by: opencode/tech-writer <tech-writer@agents.invalid>
2026-09-08 12:47:42 +02:00
28c695795a fix(M11): propagate PipelineCriticalError, redact configured secrets, signal blog failures
Correct 4 findings from the independent M11 review:

#1 (Critical) — PipelineCriticalError was downgraded to PipelineWarning:
  - Add except PipelineCriticalError: raise before each except Exception
    in all 5 non-blocking steps (fetch_blog, compare, caldav_sync, synthesis, send)
  - Critical errors now propagate to the outer handler and stop the pipeline

#2 (Critical) — redact_exception() did not use configured secrets:
  - Extend redact_exception() with extra_secrets parameter (upward compatible)
  - Harden redact_secrets(): sort extra_secrets by length descending
  - Add Settings.redaction_secrets() collecting all 6 SecretStr fields
  - Add PipelineRunner._redact(exc) using self._redaction_secrets
  - All except blocks in run() now use self._redact(exc)
  - CalDAV FAILED-status path uses full redaction_secrets collection

#3 (Medium) — BlogRSSClient silently swallowed failures:
  - Add error field to BlogRSSFetchResult
  - rss.py sets error on failure paths (except Exception, bozo/invalid feed)
  - fetch_blog_step raises RuntimeError when result.error is set
  - PipelineRunner now produces PipelineWarning for blog failures

#4 (Medium) — Test coverage at 80%, now 91%:
  - 11 new integration tests covering blog failure/success, compare failure,
    CalDAV failure (exception + FAILED status), send False/exception,
    PipelineCriticalError propagation, secret redaction with sentinel,
    empty agenda/homework, iCal cache cleanup
  - Secret redaction test uses mock (no network) and proves configured-secret
    propagation via non-URL sentinel in RuntimeError

Validation: 619 tests pass, ruff/mypy/bandit/pre-commit green, coverage 91%.

Co-authored-by: opencode/coder <coder@agents.invalid>
Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
2026-09-08 12:20:29 +02:00
26b083561a Ignore worktres 2026-09-08 11:50:25 +02:00
d7d31e14ff feat: orchestrer le pipeline M11
Co-authored-by: Codex/gpt-5.6-terra <codex-gpt-5-6-terra@agents.invalid>
2026-09-08 11:28:02 +02:00
43 changed files with 3786 additions and 148 deletions

3
.gitignore vendored
View File

@@ -52,7 +52,10 @@ Thumbs.db
# --- Local scratch / WIP files --- # --- Local scratch / WIP files ---
FIXME_* FIXME_*
FEAT_*
TEST_* TEST_*
HANDOFF.md
.worktrees/
# --- Logs --- # --- Logs ---
*.log *.log

View File

@@ -140,7 +140,7 @@
"filename": "GUIDE_DEV_PYTHON.md", "filename": "GUIDE_DEV_PYTHON.md",
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa", "hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
"is_verified": true, "is_verified": true,
"line_number": 5029, "line_number": 5064,
"is_secret": false "is_secret": false
} }
], ],
@@ -177,5 +177,5 @@
} }
] ]
}, },
"generated_at": "2026-09-08T00:29:02Z" "generated_at": "2026-09-08T10:45:46Z"
} }

View File

@@ -326,3 +326,47 @@ Un changement est considéré comme **terminé** lorsque :
- Le *handoff* distingue clairement : - Le *handoff* distingue clairement :
- Ce qui a été vérifié localement (ex. : tests unitaires, linter). - Ce qui a été vérifié localement (ex. : tests unitaires, linter).
- Ce qui nécessite encore une vérification manuelle (ex. : tests d'intégration avec un serveur CalDAV réel). - Ce qui nécessite encore une vérification manuelle (ex. : tests d'intégration avec un serveur CalDAV réel).
---
## 13. Versionnage et releases
### Politique de versionnage
Le projet suit **Semantic Versioning** (semver.org v2.0.0). Phase actuelle : `0.x` (pré-`1.0.0`).
| Changement | Incrément |
|------------|----------|
| Défaut constaté au déploiement | Patch (`0.1.Z`) — correction rétrocompatible |
| Ajout ou cassure en phase `0.x` | Minor (`0.Y.0`) |
| Déploiement réel validé | `1.0.0` |
### Règle absolue de validation
**Aucune montée de version (tag + release) ne peut être effectuée
sans validation préalable en environnement réel.** Les tests automatisés et la revue de code
ne suffisent pas ; le correctif ou la fonctionnalité doit avoir été testé avec succès
sur le serveur de production (ou un environnement équivalent) avant de tagger.
### Procédure de release
1. **Valider en environnement réel** : le correctif ou la fonctionnalité est testé
sur le serveur de production.
2. **Mettre à jour `pyproject.toml`** : incrémenter le champ `version` à la nouvelle version.
3. **Mettre à jour `CHANGELOG.md`** : ajouter une entrée sous le format
[Keep a Changelog](https://keepachangelog.com/en/1.1.0/) avec la nouvelle version et la date.
4. **Committer** : un commit `chore: monter en version x.y.z` regroupe
les mises à jour de `pyproject.toml` et `CHANGELOG.md`.
5. **Tagger** : créer un tag annoté `vx.y.z` sur le commit de version.
6. **Pousser le tag** : `git push origin vx.y.z`.
7. **Créer la release** sur Gitea avec le changelog correspondant.
### Cohérence des versions
Les trois sources de version doivent toujours être synchronisées au moment d'un tag :
- Le tag Git (`vx.y.z`)
- `pyproject.toml` (`version = "x.y.z"`)
- `CHANGELOG.md` (`## [x.y.z] - YYYY-MM-DD`)
> **Rappel** : Ne jamais créer un tag sans avoir d'abord mis à jour
> `pyproject.toml` et `CHANGELOG.md`.

28
CHANGELOG.md Normal file
View File

@@ -0,0 +1,28 @@
# Changelog
All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [0.1.0] - 2026-09-08
Initial release covering milestones M1 through M15.
### Added
- **M1 (Scaffolding)**: Python project structure with `pyproject.toml`, and tooling configuration for `ruff`, `mypy`, `bandit`, and `pre-commit`.
- **M2 (Configuration & secrets)**: Pydantic Settings for configuration management, `SecretStr` for sensitive fields, and redaction utilities (`redact_url`, `redact_secrets`, `redact_exception`) with `RedactingFormatter` for logging.
- **M3 (Data models)**: 16 Pydantic models and 6 enums across 10 modules, including frozen contracts and mutable work results.
- **M4 (Pronote sources)**: iCal fetch and parse, `pronotepy.ParentClient` integration, automatic fallback logic for `auto`, `ical`, and `pronotepy` modes, and error redaction for sensitive data.
- **M5 (Blog RSS)**: `feedparser`-based RSS client with GUID deduplication, HTTP cache support (ETag/If-Modified-Since), and `BlogRSSState` persistence.
- **M6 (Theoretical agenda)**: JSON provider with week parity (even/odd), school holidays calendar, and deterministic IDs for events.
- **M7 (CalDAV sync)**: Differential synchronization by UID, `X-PRONOTE-SYNC-MANAGED` marker for managed events, idempotent operations, preserved cancelled events, and dry-run support.
- **M8 (Agenda diff)**: `AgendaComparator` with deterministic matching, and generation of `AgendaDiff`/`AgendaChange` objects for tracking differences.
- **M9 (AI synthesis)**: `SynthesisProvider` protocol, OpenAI provider, optional `litellm` provider, and `openai-compatible` provider with degraded mode (returns `None` on failure).
- **M10 (XMPP channel)**: `XmppChannel` using `slixmpp`, formatted messages (synthesis, homeworks, changes, messages, blog), and error handling that returns `False` on failure.
- **M11 (Pipeline orchestration)**: `PipelineRunner` as composition root, 7 pipeline steps, degraded error handling, dry-run mode, and iCal reuse within a single run.
- **M12 (CLI entry point)**: `pronote-sync` command with `--dry-run` and `--log-level` options, redacted error display, and safe traceback in DEBUG mode.
- **M13 (Tests & coverage)**: 636 tests with 95.67% coverage, test fixtures (`pronote-4e.ics`, `pronote-6e.ics`), shared `conftest.py`, and secret non-leak tests.
- **M14 (Deployment)**: systemd service and timer (daily at 18:00), logrotate configuration (daily, rotate 7, compress), `check_secrets.py` pre-deployment scanner, and exploitation guide.
- **M15 (Documentation)**: README, README.LLM.md (AI agent setup guide), MIT LICENSE, CHANGELOG, and Gitea Actions CI/CD reference for LXC/VPS (Debian/CentOS).
- **Other**: MIT License. Gitea Actions CI/CD reference for LXC/VPS (Debian/CentOS) is planned and optional, not delivered in this release.

View File

@@ -1,6 +1,6 @@
# Guide de Développement : Synchronisation Pronote → CalDAV + XMPP (Python) # Guide de Développement : Synchronisation Pronote → CalDAV + XMPP (Python)
> **Statut** : Guide de référence pour un futur projet Python inspiré de [`pronote-digest`](https://github.com/antoine-coulon/pronote-digest) (TypeScript). > **Statut** : Guide de référence pour un futur projet Python inspiré de [`pronote-digest`](https://github.com/yoanbernabeu/pronote-digest) (TypeScript) par [Yoan Bernabeu](https://yoanbernabeu.github.io/pronote-digest/).
> **Public cible** : Développeurs Python (≥ 3.13.5) familiers avec les concepts de CLI, synchronisation de calendriers et messagerie instantanée. > **Public cible** : Développeurs Python (≥ 3.13.5) familiers avec les concepts de CLI, synchronisation de calendriers et messagerie instantanée.
> **Objectif** : Fournir une base architecturale et technique pour un outil **synchronisant l'agenda Pronote vers CalDAV**, **comparant avec un agenda théorique**, **récupérant messages et informations**, et **envoyant une synthèse par XMPP**. > **Objectif** : Fournir une base architecturale et technique pour un outil **synchronisant l'agenda Pronote vers CalDAV**, **comparant avec un agenda théorique**, **récupérant messages et informations**, et **envoyant une synthèse par XMPP**.
@@ -492,9 +492,12 @@ le contexte et le traceback complet.
> `redact_exception` est implémenté comme une **fonction au niveau du module** dans `utils/redaction.py`, et non comme une méthode de `RedactingFormatter` (contrairement à §4.2.2 où elle apparaît comme une méthode). > `redact_exception` est implémenté comme une **fonction au niveau du module** dans `utils/redaction.py`, et non comme une méthode de `RedactingFormatter` (contrairement à §4.2.2 où elle apparaît comme une méthode).
> `redact_url` utilise `urlsplit`/`urlunsplit`/`parse_qsl` au lieu de `urlparse`/`urlunparse`/`parse_qs`. > `redact_url` utilise `urlsplit`/`urlunsplit`/`parse_qsl` au lieu de `urlparse`/`urlunparse`/`parse_qs`.
> La correspondance des clés sensibles est insensible à la casse. > La correspondance des clés sensibles est insensible à la casse.
> `redact_secrets()` trie les `extra_secrets` par longueur décroissante pour éviter les masquages partiels.
> `Settings.redaction_secrets()` retourne un tuple des secrets configurés (mots de passe Pronote, CalDAV, XMPP et clé API IA) à passer à `redact_exception`.
```python ```python
import re import re
from typing import Iterable, SecretStr
from urllib.parse import urlparse, urlunparse, parse_qs, urlencode from urllib.parse import urlparse, urlunparse, parse_qs, urlencode
@@ -543,6 +546,20 @@ def redact_secrets(text: str) -> str:
) )
return text return text
def redact_exception(
exc: Exception, extra_secrets: Iterable[SecretStr | str] = ()
) -> str:
"""
Masque les secrets dans une exception.
:param exc: Exception à masquer.
:param extra_secrets: Secrets configurés à masquer dans le message.
:return: Message de l'exception avec les secrets masqués.
:rtype: str
"""
return redact_secrets(str(exc), extra_secrets)
``` ```
#### 4.2.2 Configuration des logs (`logging.py`) #### 4.2.2 Configuration des logs (`logging.py`)
@@ -4576,6 +4593,7 @@ class PipelineRunner:
blog_rss_client: Optional["BlogRSSClient"] = None, blog_rss_client: Optional["BlogRSSClient"] = None,
blog_state: Optional["## (section obsolète supprimée)"] = None, blog_state: Optional["## (section obsolète supprimée)"] = None,
dry_run: bool = False, dry_run: bool = False,
settings: "Settings" | None = None,
): ):
self.pronote_fetcher = pronote_fetcher self.pronote_fetcher = pronote_fetcher
self.caldav_client = caldav_client self.caldav_client = caldav_client
@@ -4587,13 +4605,19 @@ class PipelineRunner:
self.dry_run = dry_run self.dry_run = dry_run
self._errors: List[PipelineError] = [] self._errors: List[PipelineError] = []
self._warnings: List[PipelineWarning] = [] self._warnings: List[PipelineWarning] = []
self._redaction_secrets = settings.redaction_secrets() if settings else ()
def _redact(self, exc: Exception) -> str:
"""Masque les secrets configurés dans une exception."""
from ..utils.redaction import redact_exception
return redact_exception(exc, self._redaction_secrets)
def run(self) -> Tuple[Optional[PronoteData], List[PipelineError]]: def run(self) -> Tuple[Optional[PronoteData], List[PipelineError]]:
""" """
Exécute le pipeline complet. Exécute le pipeline complet.
Returns: :return: Tuple (PronoteData final, liste des erreurs).
Tuple (PronoteData final, liste des erreurs). :rtype: tuple[PronoteData | None, list[PipelineError]]
""" """
pronote_data: Optional[PronoteData] = None pronote_data: Optional[PronoteData] = None
agenda_diff = None agenda_diff = None
@@ -4622,73 +4646,81 @@ class PipelineRunner:
logger.warning(f"Étape 'normalize' échouée: {e.message}") logger.warning(f"Étape 'normalize' échouée: {e.message}")
return None, self._errors + self._warnings return None, self._errors + self._warnings
# Étape 2 bis: Récupération du blog (RSS) # Étape 2 bis: Récupération du blog (RSS)
if self.blog_rss_client and self.blog_state: if self.blog_rss_client and self.blog_state:
try: try:
blog_articles = fetch_blog_step( blog_articles = fetch_blog_step(
self.blog_rss_client, self.blog_rss_client,
self.blog_state, self.blog_state,
enabled=True, enabled=True,
) )
except PipelineError as e: except PipelineCriticalError:
self._warnings.append(PipelineWarning( raise
message=f"Récupération du blog échouée: {e.message}", except PipelineError as e:
step="fetch_blog", self._warnings.append(PipelineWarning(
)) message=f"Récupération du blog échouée: {e.message}",
logger.warning(f"Étape 'fetch_blog' échouée (non bloquante): {e.message}") step="fetch_blog",
blog_articles = [] ))
logger.warning(f"Étape 'fetch_blog' échouée (non bloquante): {e.message}")
blog_articles = []
# Étape 3: Comparaison avec l'agenda théorique # Étape 3: Comparaison avec l'agenda théorique
try: try:
agenda_diff = compare_step( agenda_diff = compare_step(
self.agenda_comparator, self.agenda_comparator,
pronote_data.lessons, pronote_data.lessons,
pronote_data.target_date, pronote_data.target_date,
) )
except PipelineError as e: except PipelineCriticalError:
self._warnings.append(PipelineWarning( raise
message=f"Comparaison échouée: {e.message}", except PipelineError as e:
step="compare", self._warnings.append(PipelineWarning(
)) message=f"Comparaison échouée: {e.message}",
logger.warning(f"Étape 'compare' échouée (non bloquante): {e.message}") step="compare",
))
logger.warning(f"Étape 'compare' échouée (non bloquante): {e.message}")
# Étape 4: Synchronisation CalDAV # Étape 4: Synchronisation CalDAV
try: try:
sync_result = caldav_sync_step( sync_result = caldav_sync_step(
self.caldav_client, self.caldav_client,
pronote_data.lessons, pronote_data.lessons,
pronote_data.homeworks, pronote_data.homeworks,
pronote_data.school_events, pronote_data.school_events,
) )
if sync_result and sync_result.status.value == "failed": if sync_result and sync_result.status.value == "failed":
self._warnings.append(PipelineWarning( self._warnings.append(PipelineWarning(
message=f"Synchronisation CalDAV échouée: {sync_result.errors}", message=f"Synchronisation CalDAV échouée: {sync_result.errors}",
step="sync", step="sync",
)) ))
logger.warning("Synchronisation CalDAV échouée (non bloquante)") logger.warning("Synchronisation CalDAV échouée (non bloquante)")
except PipelineError as e: except PipelineCriticalError:
self._warnings.append(PipelineWarning( raise
message=f"Synchronisation CalDAV échouée: {e.message}", except PipelineError as e:
step="sync", self._warnings.append(PipelineWarning(
)) message=f"Synchronisation CalDAV échouée: {e.message}",
logger.warning(f"Étape 'sync' échouée (non bloquante): {e.message}") step="sync",
))
logger.warning(f"Étape 'sync' échouée (non bloquante): {e.message}")
# Étape 5: Synthèse IA (optionnelle) # Étape 5: Synthèse IA (optionnelle)
if self.synthesis_provider and agenda_diff: if self.synthesis_provider and agenda_diff:
try: try:
synthesis_input = SynthesisInput( synthesis_input = SynthesisInput(
agenda_diff=agenda_diff, agenda_diff=agenda_diff,
messages=pronote_data.messages, messages=pronote_data.messages,
school_events=pronote_data.school_events, school_events=pronote_data.school_events,
target_date=pronote_data.target_date, target_date=pronote_data.target_date,
) )
synthesis_result = synthesis_step(self.synthesis_provider, synthesis_input) synthesis_result = synthesis_step(self.synthesis_provider, synthesis_input)
except PipelineError as e: except PipelineCriticalError:
self._warnings.append(PipelineWarning( raise
message=f"Synthèse IA échouée: {e.message}", except PipelineError as e:
step="synthesis", self._warnings.append(PipelineWarning(
)) message=f"Synthèse IA échouée: {e.message}",
logger.warning(f"Étape 'synthesis' échouée (non bloquante): {e.message}") step="synthesis",
))
logger.warning(f"Étape 'synthesis' échouée (non bloquante): {e.message}")
# Étape 6: Construction du message XMPP # Étape 6: Construction du message XMPP
xmpp_message = XmppMessage( xmpp_message = XmppMessage(
@@ -4702,29 +4734,30 @@ class PipelineRunner:
) if blog_articles else None, ) if blog_articles else None,
) )
# Étape 7: Envoi XMPP # Étape 7: Envoi XMPP
try: try:
send_step(self.channel, xmpp_message) send_step(self.channel, xmpp_message)
except PipelineError as e: except PipelineCriticalError:
self._warnings.append(PipelineWarning( raise
message=f"Envoi XMPP échoué: {e.message}", except PipelineError as e:
step="send", self._warnings.append(PipelineWarning(
)) message=f"Envoi XMPP échoué: {e.message}",
logger.warning(f"Étape 'send' échouée (non bloquante): {e.message}") step="send",
))
logger.warning(f"Étape 'send' échouée (non bloquante): {e.message}")
return pronote_data, self._errors + self._warnings return pronote_data, self._errors + self._warnings
except PipelineCriticalError as e: except PipelineCriticalError as e:
logger.error(f"Erreur critique dans le pipeline: {e.message}") logger.error(f"Erreur critique dans le pipeline: {e.message}")
return None, [e] return None, [e]
except Exception as e: except Exception as e:
from ..utils.redaction import redact_secrets safe_error = self._redact(e)
safe_error = redact_secrets(str(e)) logger.error(f"Erreur inattendue dans le pipeline: {safe_error}")
logger.error(f"Erreur inattendue dans le pipeline: {safe_error}") return None, [PipelineCriticalError(
return None, [PipelineCriticalError( message=safe_error,
message=safe_error, step="unknown",
step="unknown", )]
)]
def get_errors(self) -> List[PipelineError]: def get_errors(self) -> List[PipelineError]:
"""Récupère la liste des erreurs.""" """Récupère la liste des erreurs."""
@@ -4735,6 +4768,8 @@ class PipelineRunner:
return self._warnings return self._warnings
``` ```
Le ``PipelineRunner`` calcule ``self._redaction_secrets = settings.redaction_secrets()`` dans son constructeur. La méthode privée ``_redact(exc)`` délègue à ``redact_exception(exc, self._redaction_secrets)`` pour masquer les secrets configurés (mots de passe Pronote, CalDAV, XMPP et clé API IA). Chaque bloc ``except Exception`` utilise ``self._redact(exc)`` au lieu de ``redact_exception(exc)`` directement.
### 11.4 Étapes du pipeline (`pipeline/steps/`) ### 11.4 Étapes du pipeline (`pipeline/steps/`)
@@ -6055,7 +6090,7 @@ Ce guide fournit une **base architecturale et technique solide** pour développe
1. **Créer le dépôt** : Initialiser un nouveau dépôt Python avec la structure proposée. 1. **Créer le dépôt** : Initialiser un nouveau dépôt Python avec la structure proposée.
2. **Implémenter le cœur** : Commencer par les modules `models/`, `sources/pronote/ical.py` et `utils/`. 2. **Implémenter le cœur** : Commencer par les modules `models/`, `sources/pronote/ical.py` et `utils/`.
3. **Ajouter les tests** : Écrire des tests unitaires pour chaque module dès le début. 3. **Ajouter les tests** : Écrire des tests unitaires pour chaque module dès le début.
4. **Configurer CI/CD** : Mettre en place GitHub Actions pour exécuter les tests et vérifier la sécurité. 4. **Configurer Gitea Actions** : Mettre en place Gitea Actions pour exécuter les tests et vérifier la sécurité, en vue d'un déploiement sur LXC/VPS (Debian/CentOS).
5. **Tester en conditions réelles** : Utiliser des flux iCal Pronote anonymisés pour valider le parsing. 5. **Tester en conditions réelles** : Utiliser des flux iCal Pronote anonymisés pour valider le parsing.
> **⚠️ Rappel** : Ce guide est **volontairement détaillé** pour préserver les connaissances acquises sur les spécificités de Pronote. Certaines sections (ex: parsing iCal) contiennent des **observations précises** issues de l'analyse du code TypeScript existant. **Ne pas sous-estimer l'importance de ces détails** : ils sont critiques pour un fonctionnement fiable du projet. > **⚠️ Rappel** : Ce guide est **volontairement détaillé** pour préserver les connaissances acquises sur les spécificités de Pronote. Certaines sections (ex: parsing iCal) contiennent des **observations précises** issues de l'analyse du code TypeScript existant. **Ne pas sous-estimer l'importance de ces détails** : ils sont critiques pour un fonctionnement fiable du projet.

21
LICENSE Normal file
View File

@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2026 Antoine Van Elstraete
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

147
README.LLM.md Normal file
View File

@@ -0,0 +1,147 @@
# pronote-sync — AI Agent Setup Guide
This document guides an AI agent through installing and pre-configuring the `pronote-sync` project on a fresh Linux host (Debian/CentOS). It covers environment setup, dependency installation, and configuration file preparation. It does **NOT** cover secrets provisioning — those must be provided by the operator.
---
## Prerequisites
- Python ≥ 3.13.5 (check with `python3 --version`)
- Git
- A non-root service user (e.g., `pronote-sync`)
- Target paths:
- `/opt/pronote-sync` (code)
- `/var/lib/pronote-sync` (state)
- `/var/log/pronote-sync` (logs)
- `/etc/pronote-sync` (config)
---
## Installation Steps
```bash
# Create service user
sudo useradd --system --no-create-home --shell /usr/sbin/nologin pronote-sync
# Clone the repository
sudo git clone <repo-url> /opt/pronote-sync
sudo chown -R pronote-sync:pronote-sync /opt/pronote-sync
# Create virtual environment
cd /opt/pronote-sync
sudo -u pronote-sync python3.13 -m venv .venv
sudo -u pronote-sync .venv/bin/pip install -e ".[dev]"
# Create directories
sudo install -d -m 0700 -o pronote-sync -g pronote-sync /etc/pronote-sync
sudo install -d -m 0750 -o pronote-sync -g pronote-sync /var/lib/pronote-sync
sudo install -d -m 0750 -o pronote-sync -g pronote-sync /var/log/pronote-sync
```
---
## Configuration Preparation (Without Secrets)
```bash
# Copy the example config
sudo -u pronote-sync cp /opt/pronote-sync/.env.example /etc/pronote-sync/pronote-sync.env
# The operator must fill in secrets (PRONOTE_PASSWORD, CALDAV_PASSWORD, XMPP_PASSWORD, AI_API_KEY, etc.)
# Do NOT populate secrets automatically — leave them for the operator.
```
### Non-Secret Environment Variables (Pre-Configurable)
The following variables can be safely pre-configured in `/etc/pronote-sync/pronote-sync.env`:
- **Pronote:**
- `PRONOTE_ACCOUNT_TYPE` (default: `parent`)
- `PRONOTE_ENT` (ENT slug, e.g., `lyceeconnecte`)
- `PRONOTE_AGENDA_SOURCE`, `PRONOTE_HOMEWORK_SOURCE`, `PRONOTE_MESSAGES_SOURCE` (`auto`, `ical`, or `pronotepy`)
- **CalDAV:**
- `CALDAV_CALENDAR_PATH` (e.g., `/pronote-sync/`)
- `CALDAV_ALLOW_INSECURE_HTTP` (default: `false`)
- **Sync Window:**
- `SYNC_PAST_DAYS`, `SYNC_FUTURE_DAYS`
- **Theoretical Agenda:**
- `THEORETICAL_AGENDA_PATH`, `SCHOOL_HOLIDAYS_PATH`
- `THEORETICAL_WEEK_ANCHOR_DATE`, `THEORETICAL_WEEK_ANCHOR_TYPE`
- **XMPP:**
- `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_USE_TLS`, `XMPP_TIMEOUT`, `XMPP_RESOURCE`
- **AI:**
- `AI_ENABLED`, `AI_PROVIDER`, `AI_BASE_URL`, `AI_MODEL`, `AI_ALLOW_INSECURE_HTTP`
- **Blog:**
- `BLOG_ENABLED`, `BLOG_RSS_URL`
- **General:**
- `DRY_RUN`, `LOG_LEVEL`
### Secret Variables (Operator Must Provide)
**Do NOT set these variables automatically.** The operator must manually provide the following secrets:
- **Pronote:**
- `PRONOTE_ICAL_URL`, `PRONOTE_URL`, `PRONOTE_USERNAME`, `PRONOTE_PASSWORD`
- **CalDAV:**
- `CALDAV_URL`, `CALDAV_USERNAME`, `CALDAV_PASSWORD`
- **XMPP:**
- `XMPP_JID`, `XMPP_PASSWORD`, `XMPP_TO`
- **AI:**
- `AI_API_KEY`
---
## Pre-Deployment Checks
```bash
# Verify no secrets in the codebase
/opt/pronote-sync/.venv/bin/python /opt/pronote-sync/scripts/check_secrets.py
# Verify dependencies
/opt/pronote-sync/.venv/bin/python -m pip check
# Dry-run test (will fail without secrets, but verifies the environment)
/opt/pronote-sync/.venv/bin/pronote-sync --dry-run
```
---
## systemd Installation
```bash
# Install systemd units
sudo install -m 0644 /opt/pronote-sync/deploy/systemd/pronote-sync.service /etc/systemd/system/
sudo install -m 0644 /opt/pronote-sync/deploy/systemd/pronote-sync.timer /etc/systemd/system/
# Install logrotate config
sudo install -m 0644 /opt/pronote-sync/deploy/logrotate/pronote_sync /etc/logrotate.d/pronote_sync
# Reload and enable
sudo systemctl daemon-reload
sudo systemctl enable --now pronote-sync.timer
```
---
## Notes for the AI Agent
- **Do NOT commit or write secrets** to any file in the repository.
- **Do NOT modify** `.gitignore`, `pyproject.toml`, or existing source files.
- If `python3.13` is not available, install it first:
- Debian: `sudo apt install python3.13`
- CentOS: Compile from source or use `dnf` if available.
- The `check_secrets.py` script exits with:
- `0` (clean)
- `1` (secrets found)
- `2` (error)
- All paths in the systemd unit assume `/opt/pronote-sync` — adjust if installed elsewhere.
- The operator **must** provide real values for all **SECRET** variables before enabling the timer.

69
README.md Normal file
View File

@@ -0,0 +1,69 @@
# pronote-sync
Synchronisation Pronote → CalDAV + XMPP.
---
Synchronise l'agenda et les devoirs de **Pronote** vers un calendrier **CalDAV** et envoie un résumé quotidien par **XMPP**. Supporte les sources iCal et `pronotepy` avec repli automatique. Synthèse IA optionnelle.
---
## 🚀 Démarrage rapide
```bash
# Cloner le dépôt
git clone <repo-url>
cd pronote-sync
# Créer l'environnement virtuel
python3.13 -m venv .venv
source .venv/bin/activate
# Installer
pip install -e ".[dev]"
# Configurer
cp .env.example .env
# Éditer .env avec vos paramètres (voir .env.example pour le détail)
# Tester
pronote-sync --dry-run --log-level DEBUG
```
---
## 📖 Utilisation
```bash
pronote-sync # Exécute la synchronisation
pronote-sync --dry-run # Simulation sans écriture
pronote-sync --log-level DEBUG # Verbosité des journaux
```
---
## 🛠️ Déploiement
Les artefacts pour **systemd/timer** et **logrotate** sont fournis dans `deploy/`. Voir [docs/exploitation.md](docs/exploitation.md) pour plus de détails.
---
## 🙏 Remerciements
Ce projet repose sur les bibliothèques open-source suivantes :
- [pronotepy](https://github.com/bain3/pronotepy) — client Pronote
- [icalendar](https://github.com/collective/icalendar) — parsing iCal
- [caldav](https://github.com/python-caldav/caldav) — client CalDAV
- [slixmpp](https://github.com/poezio/slixmpp) — client XMPP
- [pydantic](https://github.com/pydantic/pydantic) — validation et configuration
- [openai](https://github.com/openai/openai-python) — synthèse IA
- [feedparser](https://github.com/kurtmckee/feedparser) — parsing RSS
- [beautifulsoup4](https://www.crummy.com/software/BeautifulSoup/) — parsing HTML
Inspiré de [pronote-digest](https://github.com/yoanbernabeu/pronote-digest) par [Yoan Bernabeu](https://yoanbernabeu.github.io/pronote-digest/).
---
## Licence
MIT — voir [LICENSE](LICENSE).

59
TODO.md
View File

@@ -219,20 +219,21 @@ Construire et envoyer le message XMPP structuré via un compte bot dédié (mess
Composer et orchestrer toutes les étapes avec gestion d'erreurs dégradée et mode dry-run. Composer et orchestrer toutes les étapes avec gestion d'erreurs dégradée et mode dry-run.
- [ ] Compléter si nécessaire la hiérarchie canonique dans `pronote_sync/errors.py` (`ErrorSeverity`, `PipelineError`, `PipelineWarning`, `PipelineCriticalError`) ; ne pas créer de doublon dans `pipeline/steps/errors.py`. - [x] Compléter si nécessaire la hiérarchie canonique dans `pronote_sync/errors.py` (`ErrorSeverity`, `PipelineError`, `PipelineWarning`, `PipelineCriticalError`) ; ne pas créer de doublon dans `pipeline/steps/errors.py`.
- [ ] Créer les étapes `pipeline/steps/` : `fetch.py`, `normalize.py`, `compare.py`, `caldav_sync.py`, `synthesis.py`, `send.py`, `fetch_blog.py`. - [x] Créer les étapes `pipeline/steps/` : `fetch.py`, `normalize.py`, `compare.py`, `caldav_sync.py`, `synthesis.py`, `send.py`, `fetch_blog.py`.
- [ ] Créer `pipeline/run.py` : `PipelineRunner` (composition root) orchestrant fetch → normalize → fetch_blog → compare → caldav_sync → synthesis → send. - [x] Créer `pipeline/run.py` : `PipelineRunner` (composition root) orchestrant fetch → normalize → fetch_blog → compare → caldav_sync → synthesis → send.
- [ ] Gérer les erreurs dégradées (continuer sauf critique) et renvoyer `(PronoteData, erreurs + warns)`. - [x] Gérer les erreurs dégradées (continuer sauf critique) et renvoyer `(PronoteData, erreurs + warns)`.
- [ ] Implémenter le mode `dry_run` (aucune écriture CalDAV/XMPP). - [x] Implémenter le mode `dry_run` (aucune écriture CalDAV/XMPP).
- [ ] Câbler l'injection des dépendances (Protocol + composition root), sans singleton global. - [x] Câbler l'injection des dépendances (Protocol + composition root), sans singleton global.
- [ ] Réutiliser, dans une même exécution, un unique téléchargement/parsing iCal pour l'agenda et les devoirs lorsque les sources sélectionnées le permettent ; rester sur un cache local au run, sans cache global ni persistant. - [x] Réutiliser, dans une même exécution, un unique téléchargement/parsing iCal pour l'agenda et les devoirs lorsque les sources sélectionnées le permettent ; rester sur un cache local au run, sans cache global ni persistant.
### Critères d'acceptation ### Critères d'acceptation
- Le pipeline complet s'exécute de bout en bout (mocks) dans le bon ordre. - [x] Le pipeline complet s'exécute de bout en bout (mocks) dans le bon ordre.
- Une sélection iCal commune à l'agenda et aux devoirs ne déclenche qu'un téléchargement/parsing du flux par run. - [x] Une sélection iCal commune à l'agenda et aux devoirs ne déclenche qu'un téléchargement/parsing du flux par run.
- Une erreur non critique (ex : synthèse IA) n'empêche pas l'envoi XMPP. - [x] Une erreur non critique (ex : synthèse IA) n'empêche pas l'envoi XMPP.
- `dry_run=True` n'effectue aucune écriture ; aucune source disponible → erreur critique explicite. - [x] `dry_run=True` n'effectue aucune écriture ; aucune source disponible → erreur critique explicite.
- Si `THEORETICAL_AGENDA_PATH` est absent, le pipeline produit un diff vide sans erreur et n'instancie pas `AgendaComparator` ; si présent, il instancie le comparateur et effectue la comparaison. - [x] Si `THEORETICAL_AGENDA_PATH` est absent, le pipeline produit un diff vide sans erreur et n'instancie pas `AgendaComparator` ; si présent, il instancie le comparateur et effectue la comparaison.
- [x] Les erreurs critiques (`PipelineCriticalError`) propagées depuis une étape non-bloquante arrêtent le pipeline.
--- ---
@@ -240,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.
@@ -256,8 +257,8 @@ Exposer le lancement du pipeline via une interface en ligne de commande.
Couvrir l'ensemble du code par des tests sans réseau, avec fixtures anonymisées, jusqu'à ≥ 90 %. Couvrir l'ensemble du code par des tests sans réseau, avec fixtures anonymisées, jusqu'à ≥ 90 %.
- [ ] Créer `tests/fixtures/` : `pronote-4e.ics`, `pronote-6e.ics`, `theoretical.json`, `school_holidays.json`, `blog_rss.xml` (anonymisés, sans `icalsecurise`). - [x] Créer `tests/fixtures/` : `pronote-4e.ics`, `pronote-6e.ics`, `theoretical.json`, `school_holidays.json`, `blog_rss.xml` (anonymisés, sans `icalsecurise`).
- [ ] Créer `tests/conftest.py` : fixtures partagées (sample_lesson, sample_cancelled_lesson, sample_homework, sample_school_event, sample_message, sample_pronote_data…). - [x] Créer `tests/conftest.py` : fixtures partagées (sample_lesson, sample_cancelled_lesson, sample_homework, sample_school_event, sample_message, sample_pronote_data…).
- [x] Écrire `tests/unit/` : `test_models`, `test_parsing` (iCal), `test_uid`, `test_redaction`, `test_diff`, `test_sync`. - [x] Écrire `tests/unit/` : `test_models`, `test_parsing` (iCal), `test_uid`, `test_redaction`, `test_diff`, `test_sync`.
- [x] Couvrir les régressions M4 : signature réelle de `ParentClient`, ENT autorisé/inconnu, erreur vs résultat vide, `STATUS:CANCELLED` sans catégorie, plusieurs devoirs à la même date, filtrage `pronotepy` sur la date cible et stabilité d'identité entre sources. - [x] Couvrir les régressions M4 : signature réelle de `ParentClient`, ENT autorisé/inconnu, erreur vs résultat vide, `STATUS:CANCELLED` sans catégorie, plusieurs devoirs à la même date, filtrage `pronotepy` sur la date cible et stabilité d'identité entre sources.
- [x] Écrire `tests/integration/` : `test_pipeline`, `test_caldav` (mocké), `test_xmpp` (mocké). - [x] Écrire `tests/integration/` : `test_pipeline`, `test_caldav` (mocké), `test_xmpp` (mocké).
@@ -276,11 +277,11 @@ Couvrir l'ensemble du code par des tests sans réseau, avec fixtures anonymisée
Mettre en production de façon supervisée (planification, rotation des logs, vérification des secrets). Mettre en production de façon supervisée (planification, rotation des logs, vérification des secrets).
- [ ] Créer une unité systemd (`pronote-sync.service` + timer) ou une ligne cron (exécution quotidienne). - [x] Créer une unité systemd (`pronote-sync.service` + timer) ou une ligne cron (exécution quotidienne).
- [ ] Créer `logrotate.d/pronote_sync` (daily, rotate 7, compress, delaycompress). - [x] Créer `logrotate.d/pronote_sync` (daily, rotate 7, compress, delaycompress).
- [ ] Ajouter un script de vérification des secrets (§13.6) exécuté avant chaque déploiement. - [x] Ajouter un script de vérification des secrets (§13.6) exécuté avant chaque déploiement.
- [ ] Documenter la supervision (logs, alertes en cas d'échec) et la maintenance (maj dépendances, dry-run avant MAJ). - [x] Documenter la supervision (logs, alertes en cas d'échec) et la maintenance (maj dépendances, dry-run avant MAJ).
- [ ] Vérifier `pip check` et tester le dry-run avant mise en production. - [x] Vérifier `pip check` et tester le dry-run avant mise en production.
### Critères d'acceptation ### Critères d'acceptation
- Le service/timer systemd (ou cron) lance le pipeline quotidiennement. - Le service/timer systemd (ou cron) lance le pipeline quotidiennement.
@@ -293,13 +294,13 @@ Mettre en production de façon supervisée (planification, rotation des logs, v
Rédiger la documentation utilisateur et finaliser le projet. Rédiger la documentation utilisateur et finaliser le projet.
- [ ] Créer `README.md` (installation, configuration `.env`, usage CLI, systemd/docker, limites, RGPD). - [x] Créer `README.md` (installation, configuration `.env`, usage CLI, systemd/docker, limites, RGPD).
- [ ] Documenter l'architecture (pipeline, modules) en résumé. - [x] Documenter l'architecture (pipeline, modules) en résumé.
- [ ] Ajouter `CHANGELOG` initial et la licence (MIT). - [x] Ajouter `CHANGELOG` initial et la licence (MIT).
- [ ] Revue finale : cohérence avec le guide, aucun secret documenté en clair. - [x] Revue finale : cohérence avec le guide, aucun secret documenté en clair.
- [ ] (Optionnel) Configurer GitHub Actions CI/CD (pytest + bandit + ruff + mypy) d'après §Prochaines étapes. - [ ] (Optionnel) Configurer Gitea Actions (pytest + bandit + ruff + mypy) pour le déploiement LXC/VPS (Debian/CentOS).
### Critères d'acceptation ### Critères d'acceptation
- `README.md` permet d'installer et de lancer le projet sans le guide. - `README.md` permet d'installer et de lancer le projet sans le guide.
- La CI exécute tests + lint + sécurité. - Gitea Actions exécute tests + lint + sécurité.
- Aucun secret dans la documentation. - Aucun secret dans la documentation.

31
data/school_holidays.json Normal file
View File

@@ -0,0 +1,31 @@
{
"zone": "A",
"school_year": "2026-2027",
"periods": [
{
"start_date": "2026-10-17",
"end_date": "2026-11-02",
"label": "Toussaint"
},
{
"start_date": "2026-12-19",
"end_date": "2027-01-04",
"label": "Noël"
},
{
"start_date": "2027-02-13",
"end_date": "2027-03-01",
"label": "Hiver"
},
{
"start_date": "2027-04-10",
"end_date": "2027-04-26",
"label": "Printemps"
},
{
"start_date": "2027-07-03",
"end_date": "2027-09-01",
"label": "Été"
}
]
}

View File

@@ -0,0 +1,9 @@
/var/log/pronote-sync/pronote-sync.log {
daily
missingok
rotate 7
compress
delaycompress
notifempty
create 0640 pronote-sync pronote-sync
}

View File

@@ -0,0 +1,23 @@
[Unit]
Description=Synchronisation Pronote vers CalDAV et XMPP
Wants=network-online.target
After=network-online.target
[Service]
Type=oneshot
User=pronote-sync
Group=pronote-sync
WorkingDirectory=/var/lib/pronote-sync
EnvironmentFile=/etc/pronote-sync/pronote-sync.env
Environment=PYTHONUNBUFFERED=1
StateDirectory=pronote-sync
LogsDirectory=pronote-sync
ExecStartPre=/opt/pronote-sync/.venv/bin/python /opt/pronote-sync/scripts/check_secrets.py
ExecStart=/opt/pronote-sync/.venv/bin/pronote-sync
StandardOutput=append:/var/log/pronote-sync/pronote-sync.log
StandardError=append:/var/log/pronote-sync/pronote-sync.log
NoNewPrivileges=true
PrivateTmp=true
ProtectHome=true
ProtectSystem=strict
ReadWritePaths=/var/lib/pronote-sync /var/log/pronote-sync

View File

@@ -0,0 +1,10 @@
[Unit]
Description=Exécution quotidienne de pronote-sync
[Timer]
OnCalendar=*-*-* 18:00:00
Persistent=true
Unit=pronote-sync.service
[Install]
WantedBy=timers.target

143
docs/exploitation.md Normal file
View File

@@ -0,0 +1,143 @@
# Exploitation de `pronote-sync`
Ce guide décrit l'installation et l'exploitation des artefacts de déploiement
fournis par le projet. Les paramètres de l'unité systemd fournie sont des
exemples d'installation : adaptez-les à l'hôte cible avant son installation.
Ne placez jamais de secret dans une unité systemd, une commande shell, un
journal ou ce document.
## Préparer l'hôte
Installez le projet et ses dépendances dans le répertoire choisi, puis créez le
fichier d'environnement référencé par l'unité à partir de `.env.example`. Il
doit rester local et lisible uniquement par le compte de service :
```bash
sudo install -d -m 0700 -o <utilisateur-service> -g <groupe-service> <repertoire-configuration>
sudo install -m 0600 -o <utilisateur-service> -g <groupe-service> .env <fichier-environnement>
```
Les unités fournies nécessitent l'interface CLI livrée au jalon M12. Avant de
les installer, vérifiez que la version installée contient bien ce point
d'entrée :
```bash
.venv/bin/pronote-sync --help
```
Avant toute activation ou mise à jour, exécutez les contrôles depuis la racine
du projet :
```bash
.venv/bin/python scripts/check_secrets.py
.venv/bin/python -m pip check
.venv/bin/pronote-sync --dry-run
```
Le contrôle des secrets doit réussir avant le déploiement. Il inspecte les
fichiers textuels de l'artefact, en excluant volontairement `.env`, les
environnements virtuels, les répertoires générés, `tests/` et
`GUIDE_DEV_PYTHON.md` ; les sentinelles et exemples de ces deux derniers ne
bloquent donc pas le déploiement. Il ne valide ni les valeurs ni les permissions
du fichier d'environnement. Pour analyser seulement le contenu indexé avant un
commit, utilisez `scripts/check_secrets.py --staged`.
Le dry-run vérifie le pipeline sans appliquer les écritures de synchronisation ;
il ne remplace pas une vérification des paramètres réellement chargés.
## Installation systemd
Les fichiers versionnés sont :
- `deploy/systemd/pronote-sync.service` ;
- `deploy/systemd/pronote-sync.timer`.
Copiez-les dans le répertoire d'unités systemd de l'hôte. Avant de les activer,
adaptez `User`, `Group`, `WorkingDirectory`, `EnvironmentFile`, les chemins des
exécutables dans `ExecStartPre` et `ExecStart`, ainsi que les chemins de
`StateDirectory`, `LogsDirectory` et `ReadWritePaths`. L'artefact fourni prend
pour exemple le compte `pronote-sync`, le code dans `/opt/pronote-sync`, l'état
dans `/var/lib/pronote-sync`, les logs dans `/var/log/pronote-sync` et le fichier
d'environnement `/etc/pronote-sync/pronote-sync.env`. Ne copiez pas de valeur
secrète dans l'unité.
```bash
sudo install -m 0644 deploy/systemd/pronote-sync.service /etc/systemd/system/
sudo install -m 0644 deploy/systemd/pronote-sync.timer /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable --now pronote-sync.timer
systemctl list-timers pronote-sync.timer
```
Pour tester une exécution sans attendre la prochaine échéance :
```bash
sudo systemctl start pronote-sync.service
sudo systemctl status pronote-sync.service
```
Une exécution en échec laisse l'unité `pronote-sync.service` en état `failed`.
La supervision de l'hôte doit donc déclencher une alerte sur cet état ou sur un
échec du timer/service ; le transport de cette alerte (courriel, XMPP ou système
de supervision) relève de l'exploitation locale.
## Journaux et alertes
La configuration systemd redirige la sortie standard et la sortie d'erreur vers
`/var/log/pronote-sync/pronote-sync.log`. Consultez ce fichier ou, selon la
configuration de l'hôte, le journal de l'unité :
```bash
sudo tail -f /var/log/pronote-sync/pronote-sync.log
sudo journalctl -u pronote-sync.service --since today
sudo journalctl -u pronote-sync.service -f
systemctl status pronote-sync.timer
```
Traitez un statut non nul ou une unité `failed` comme un échec à investiguer.
Les logs applicatifs masquent les secrets configurés, mais évitez tout de même
de partager sans relecture un export de journal : une donnée sensible issue de
l'environnement ou d'un outil tiers ne doit pas être supposée sûre par défaut.
## Rotation des journaux
L'artefact `deploy/logrotate/pronote_sync` cible le fichier
`/var/log/pronote-sync/pronote-sync.log` utilisé par l'unité fournie. Installez-
le puis validez sa syntaxe avant activation :
```bash
sudo install -m 0644 deploy/logrotate/pronote_sync /etc/logrotate.d/pronote_sync
sudo logrotate --debug /etc/logrotate.d/pronote_sync
```
La rotation configurée est quotidienne, conserve sept archives et utilise
`compress` avec `delaycompress`. Elle recrée le fichier avec les droits `0640`
pour le compte de service. Si vous modifiez le chemin de journal dans l'unité,
mettez aussi à jour la règle logrotate correspondante.
## Mise à jour et retour au service
Avant de remplacer les dépendances ou le code, conservez une copie protégée du
fichier d'environnement local, sans l'ajouter au dépôt. Après la mise à jour,
réexécutez, dans cet ordre, les contrôles de secrets, de cohérence des paquets
et le dry-run :
```bash
.venv/bin/python scripts/check_secrets.py
.venv/bin/python -m pip check
.venv/bin/pronote-sync --dry-run
```
Rechargez ensuite les unités si leurs fichiers ont changé, puis vérifiez une
exécution et son journal :
```bash
sudo systemctl daemon-reload
sudo systemctl restart pronote-sync.timer
sudo systemctl start pronote-sync.service
journalctl -u pronote-sync.service -n 100 --no-pager
```
En cas d'échec, ne relancez pas automatiquement après avoir modifié des
identifiants : corrigez la configuration locale, repassez le contrôle des
secrets et le dry-run, puis consultez le journal expurgé.

View File

@@ -0,0 +1 @@
"""Interface en ligne de commande du pipeline ``pronote-sync``."""

153
pronote_sync/cli/main.py Normal file
View 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())

View File

@@ -261,3 +261,23 @@ class Settings(BaseSettings):
ai: AISettings = Field(default_factory=AISettings) ai: AISettings = Field(default_factory=AISettings)
blog: BlogSettings = Field(default_factory=BlogSettings) blog: BlogSettings = Field(default_factory=BlogSettings)
app: AppSettings = Field(default_factory=AppSettings) app: AppSettings = Field(default_factory=AppSettings)
def redaction_secrets(self) -> tuple[SecretStr, ...]:
"""Énumère tous les secrets configurés pour la rédaction.
Collecte les valeurs :class:`pydantic.SecretStr` non vides présentes
dans les sous-configurations (Pronote, CalDAV, XMPP, IA). Les valeurs
vides ou ``None`` sont filtrées ; les doublons sont supprimés.
:return: Tuple de secrets à masquer dans les messages d'erreur.
:rtype: tuple[SecretStr, ...]
"""
secrets = [
self.pronote.ical_url,
self.pronote.password,
self.caldav.url,
self.caldav.password,
self.xmpp.password,
self.ai.api_key,
]
return tuple(dict.fromkeys(secret for secret in secrets if secret is not None))

View File

@@ -2,6 +2,8 @@
from __future__ import annotations from __future__ import annotations
from enum import StrEnum
class PronoteSyncError(Exception): class PronoteSyncError(Exception):
"""Erreur de base pour toutes les exceptions du projet pronote-sync. """Erreur de base pour toutes les exceptions du projet pronote-sync.
@@ -16,24 +18,67 @@ class PronoteSyncError(Exception):
:param message: Message décrivant la cause de l'erreur. :param message: Message décrivant la cause de l'erreur.
""" """
super().__init__(message) super().__init__(message)
self.message = message
class PipelineCriticalError(PronoteSyncError): class ErrorSeverity(StrEnum):
"""Niveau de gravité d'une erreur produite par le pipeline."""
WARNING = "warning"
CRITICAL = "critical"
class PipelineError(PronoteSyncError):
"""Erreur structurée produite par une étape du pipeline.
:ivar severity: Niveau de gravité de l'erreur.
:ivar step: Étape ayant produit l'erreur, si elle est connue.
:ivar recoverable: Indique si le pipeline peut poursuivre son exécution.
"""
def __init__(
self,
message: str,
*,
severity: ErrorSeverity = ErrorSeverity.WARNING,
step: str | None = None,
recoverable: bool = True,
) -> None:
"""Initialise une erreur de pipeline.
:param message: Message descriptif expurgé.
:param severity: Niveau de gravité associé.
:param step: Étape ayant produit l'erreur.
:param recoverable: ``True`` si le pipeline peut continuer.
"""
super().__init__(message)
self.severity = severity
self.step = step
self.recoverable = recoverable
class PipelineCriticalError(PipelineError):
"""Erreur critique du pipeline, levée quand aucune récupération n'est possible. """Erreur critique du pipeline, levée quand aucune récupération n'est possible.
Par exemple : échec simultané des sources iCal et pronotepy, Par exemple : échec simultané des sources iCal et pronotepy,
rendant impossible toute synchronisation. rendant impossible toute synchronisation.
""" """
def __init__(self, message: str) -> None: def __init__(self, message: str, step: str | None = None) -> None:
"""Initialise l'erreur critique avec un message descriptif. """Initialise l'erreur critique avec un message descriptif.
:param message: Message décrivant la cause de l'erreur critique. :param message: Message décrivant la cause de l'erreur critique.
:param step: Étape ayant produit l'erreur critique.
""" """
super().__init__(message) super().__init__(
message,
severity=ErrorSeverity.CRITICAL,
step=step,
recoverable=False,
)
class PipelineWarning(PronoteSyncError): class PipelineWarning(PipelineError):
"""Avertissement non bloquant pour une erreur récupérable du pipeline. """Avertissement non bloquant pour une erreur récupérable du pipeline.
Contrairement à :class:`PipelineCriticalError`, cet avertissement signale Contrairement à :class:`PipelineCriticalError`, cet avertissement signale
@@ -54,6 +99,9 @@ class PipelineWarning(PronoteSyncError):
:param message: Message décrivant la cause de l'avertissement. :param message: Message décrivant la cause de l'avertissement.
:param step: Étape du pipeline ayant produit l'avertissement. :param step: Étape du pipeline ayant produit l'avertissement.
""" """
super().__init__(message) super().__init__(
self.recoverable = True message,
self.step = step severity=ErrorSeverity.WARNING,
step=step,
recoverable=True,
)

View File

@@ -0,0 +1,5 @@
"""Orchestration du pipeline Pronote → CalDAV → XMPP."""
from pronote_sync.pipeline.run import PipelineRunner
__all__ = ["PipelineRunner"]

View File

@@ -0,0 +1,299 @@
"""Composition root et orchestrateur du pipeline Pronote → CalDAV → XMPP."""
from __future__ import annotations
import logging
from collections.abc import Callable
from contextlib import AbstractContextManager, nullcontext
from datetime import datetime
from typing import Protocol, runtime_checkable
from pronote_sync.channels import get_channel
from pronote_sync.channels.protocol import Channel
from pronote_sync.config.settings import Settings
from pronote_sync.errors import PipelineCriticalError, PipelineError, PipelineWarning
from pronote_sync.models.blog import ExternalInfo
from pronote_sync.models.pronote import PronoteData
from pronote_sync.models.sync import CalDAVSyncResult, CalDAVSyncStatus
from pronote_sync.models.synthesis import SynthesisInput
from pronote_sync.models.xmpp import XmppMessage
from pronote_sync.pipeline.steps.caldav_sync import CalDAVSynchronizer, caldav_sync_step
from pronote_sync.pipeline.steps.compare import compare_step
from pronote_sync.pipeline.steps.fetch import fetch_step
from pronote_sync.pipeline.steps.fetch_blog import fetch_blog_step
from pronote_sync.pipeline.steps.normalize import normalize_step
from pronote_sync.pipeline.steps.send import send_step
from pronote_sync.pipeline.steps.synthesis import synthesis_step
from pronote_sync.sources.blog.rss import BlogRSSClient
from pronote_sync.sources.blog.state import BlogRSSState
from pronote_sync.sources.pronote.client import PronoteClient
from pronote_sync.sources.pronote.fallback import PronoteFetcher, PronoteFetcherProtocol
from pronote_sync.sources.theoretical import get_theoretical_provider
from pronote_sync.sync.diff import AgendaComparator
from pronote_sync.sync.synchronizer import synchronize
from pronote_sync.synthesis import get_synthesis_provider
from pronote_sync.synthesis.provider import SynthesisProvider
from pronote_sync.utils.redaction import redact_exception, redact_secrets
logger = logging.getLogger(__name__)
def _synchronize_caldav(data: PronoteData, settings: Settings) -> CalDAVSyncResult:
"""Adapte le synchroniseur CalDAV de production au protocole injecté.
:param data: Données Pronote normalisées à synchroniser.
:param settings: Configuration effective de l'exécution.
:return: Résultat de la synchronisation CalDAV.
:rtype: CalDAVSyncResult
"""
return synchronize(data, settings)
@runtime_checkable
class _RunContextFetcher(PronoteFetcherProtocol, Protocol):
"""Protocole interne d'un fetcher capable d'isoler un cache par run."""
def run_context(self) -> AbstractContextManager[None]:
"""Retourne le contexte de durée de vie d'une exécution.
:return: Contexte éphémère associé à l'exécution.
:rtype: AbstractContextManager[None]
"""
...
class PipelineRunner:
"""Orchestre les étapes fetch → normalize → blog → compare → CalDAV → IA → XMPP.
Toutes les dépendances sont injectables. La méthode :meth:`from_settings`
constitue la composition root de production et ne crée aucun singleton.
"""
def __init__(
self,
*,
settings: Settings,
pronote_fetcher: PronoteFetcherProtocol,
caldav_synchronizer: CalDAVSynchronizer = _synchronize_caldav,
agenda_comparator: AgendaComparator | None = None,
synthesis_provider: SynthesisProvider | None = None,
channel: Channel | None = None,
blog_client: BlogRSSClient | None = None,
blog_state: BlogRSSState | None = None,
dry_run: bool | None = None,
now_provider: Callable[[], datetime] = datetime.now,
) -> None:
"""Initialise un pipeline entièrement injectable.
:param settings: Configuration de base du pipeline.
:param pronote_fetcher: Source Pronote à utiliser.
:param caldav_synchronizer: Service CalDAV injecté.
:param agenda_comparator: Comparateur théorique, absent si désactivé.
:param synthesis_provider: Fournisseur IA optionnel.
:param channel: Canal XMPP optionnel.
:param blog_client: Client RSS optionnel.
:param blog_state: État RSS associé au client optionnel.
:param dry_run: Surcharge optionnelle du mode dry-run de la configuration.
:param now_provider: Horloge injectée pour rendre l'exécution testable.
"""
self._settings = settings
self._redaction_secrets = settings.redaction_secrets()
self._pronote_fetcher = pronote_fetcher
self._caldav_synchronizer = caldav_synchronizer
self._agenda_comparator = agenda_comparator
self._synthesis_provider = synthesis_provider
self._channel = channel
self._blog_client = blog_client
self._blog_state = blog_state
self._dry_run = settings.app.dry_run if dry_run is None else dry_run
self._now_provider = now_provider
self._errors: list[PipelineError] = []
self._warnings: list[PipelineWarning] = []
@classmethod
def from_settings(cls, settings: Settings, *, dry_run: bool | None = None) -> PipelineRunner:
"""Construit les dépendances de production sans singleton global.
:param settings: Configuration validée de l'application.
:param dry_run: Surcharge optionnelle du mode dry-run.
:return: Pipeline prêt à être exécuté.
:rtype: PipelineRunner
"""
effective_dry_run = settings.app.dry_run if dry_run is None else dry_run
theoretical_provider = get_theoretical_provider(
settings.app.theoretical_agenda_path,
settings.app.school_holidays_path,
settings.app.theoretical_week_anchor_date,
settings.app.theoretical_week_anchor_type,
)
comparator = (
AgendaComparator(theoretical_provider) if theoretical_provider is not None else None
)
blog_client = BlogRSSClient(settings.blog.rss_url) if settings.blog.enabled else None
blog_state = BlogRSSState() if settings.blog.enabled else None
return cls(
settings=settings,
pronote_fetcher=PronoteFetcher(settings, PronoteClient(settings.pronote)),
agenda_comparator=comparator,
synthesis_provider=get_synthesis_provider(settings.ai),
channel=get_channel(settings.xmpp, dry_run=effective_dry_run),
blog_client=blog_client,
blog_state=blog_state,
dry_run=effective_dry_run,
)
def _effective_settings(self) -> Settings:
"""Retourne la configuration dont le dry-run reflète l'exécution courante.
:return: Copie de configuration à passer aux dépendances.
:rtype: Settings
"""
if self._settings.app.dry_run == self._dry_run:
return self._settings
return self._settings.model_copy(
update={"app": self._settings.app.model_copy(update={"dry_run": self._dry_run})}
)
def _redact(self, exc: Exception) -> str:
"""Rédige une exception avec les secrets configurés.
:param exc: Exception dont le message doit être masqué.
:return: Message d'erreur avec secrets configurés remplacés par ``REDACTED``.
:rtype: str
"""
return redact_exception(exc, self._redaction_secrets)
def _run_context(self) -> AbstractContextManager[None]:
"""Retourne le contexte isolant les éventuels caches de source.
:return: Contexte de durée de vie du run, vide pour un fetcher générique.
:rtype: AbstractContextManager[None]
"""
if isinstance(self._pronote_fetcher, _RunContextFetcher):
return self._pronote_fetcher.run_context()
return nullcontext()
def _warn(self, step: str, message: str) -> None:
"""Enregistre et journalise un avertissement expurgé.
:param step: Étape ayant échoué.
:param message: Message déjà expurgé.
"""
warning = PipelineWarning(message, step=step)
self._warnings.append(warning)
logger.warning("Étape %s dégradée : %s", step, warning.message)
def run(self) -> tuple[PronoteData | None, list[PipelineError]]:
"""Exécute le pipeline complet dans l'ordre contractuel.
Une erreur de récupération critique interrompt l'exécution. Les erreurs
des étapes facultatives sont converties en :class:`PipelineWarning` afin
que les étapes suivantes, notamment XMPP, restent exécutées.
:return: Données Pronote normalisées ou ``None``, puis erreurs et avertissements.
:rtype: tuple[PronoteData | None, list[PipelineError]]
"""
self._errors = []
self._warnings = []
now = self._now_provider()
effective_settings = self._effective_settings()
try:
with self._run_context():
fetched, fetch_warnings = fetch_step(self._pronote_fetcher, today=now.date())
self._warnings.extend(fetch_warnings)
data = normalize_step(fetched, generated_at=now)
try:
blog_articles = fetch_blog_step(self._blog_client, self._blog_state)
except PipelineCriticalError:
raise
except Exception as exc:
self._warn("fetch_blog", self._redact(exc))
blog_articles = []
try:
agenda_diff = compare_step(self._agenda_comparator, data)
except PipelineCriticalError:
raise
except Exception as exc:
self._warn("compare", self._redact(exc))
from pronote_sync.models.diff import AgendaDiff
agenda_diff = AgendaDiff(target_date=data.target_date)
try:
sync_result = caldav_sync_step(
self._caldav_synchronizer, data, effective_settings
)
if sync_result.status is CalDAVSyncStatus.FAILED:
caldav_errors = redact_secrets(
"; ".join(sync_result.errors),
extra_secrets=self._redaction_secrets,
)
self._warn("caldav_sync", caldav_errors or "Échec CalDAV")
except PipelineCriticalError:
raise
except Exception as exc:
self._warn("caldav_sync", self._redact(exc))
try:
synthesis = synthesis_step(
self._synthesis_provider,
SynthesisInput(
agenda_diff=agenda_diff,
messages=data.messages,
school_events=data.school_events,
target_date=data.target_date,
),
)
except PipelineCriticalError:
raise
except Exception as exc:
self._warn("synthesis", self._redact(exc))
synthesis = None
message = XmppMessage(
target_date=data.target_date,
synthesis=synthesis.text if synthesis is not None else None,
homeworks=tuple(data.homeworks),
changes=agenda_diff.changes,
messages=tuple(data.messages),
external_info=ExternalInfo(blog_articles=tuple(blog_articles))
if blog_articles
else None,
)
if self._channel is not None and not self._dry_run:
try:
if not send_step(self._channel, message):
self._warn("send", "Le canal XMPP a refusé l'envoi")
except PipelineCriticalError:
raise
except Exception as exc:
self._warn("send", self._redact(exc))
return data, [*self._errors, *self._warnings]
except PipelineCriticalError as exc:
logger.error("Erreur critique du pipeline : %s", exc.message)
self._errors.append(exc)
except Exception as exc:
error = PipelineCriticalError(
f"Erreur inattendue du pipeline : {self._redact(exc)}", step="pipeline"
)
logger.error("Erreur critique du pipeline : %s", error.message)
self._errors.append(error)
return None, [*self._errors, *self._warnings]
def get_errors(self) -> list[PipelineError]:
"""Retourne les erreurs critiques de la dernière exécution.
:return: Copie des erreurs critiques.
:rtype: list[PipelineError]
"""
return list(self._errors)
def get_warnings(self) -> list[PipelineWarning]:
"""Retourne les avertissements de la dernière exécution.
:return: Copie des avertissements non bloquants.
:rtype: list[PipelineWarning]
"""
return list(self._warnings)

View File

@@ -0,0 +1,19 @@
"""Étapes isolées utilisées par l'orchestrateur du pipeline."""
from pronote_sync.pipeline.steps.caldav_sync import caldav_sync_step
from pronote_sync.pipeline.steps.compare import compare_step
from pronote_sync.pipeline.steps.fetch import fetch_step
from pronote_sync.pipeline.steps.fetch_blog import fetch_blog_step
from pronote_sync.pipeline.steps.normalize import normalize_step
from pronote_sync.pipeline.steps.send import send_step
from pronote_sync.pipeline.steps.synthesis import synthesis_step
__all__ = [
"caldav_sync_step",
"compare_step",
"fetch_blog_step",
"fetch_step",
"normalize_step",
"send_step",
"synthesis_step",
]

View File

@@ -0,0 +1,37 @@
"""Étape d'appel à la synchronisation CalDAV."""
from __future__ import annotations
from typing import Protocol
from pronote_sync.config.settings import Settings
from pronote_sync.models.pronote import PronoteData
from pronote_sync.models.sync import CalDAVSyncResult
class CalDAVSynchronizer(Protocol):
"""Protocole injectable de synchronisation CalDAV."""
def __call__(self, data: PronoteData, settings: Settings) -> CalDAVSyncResult:
"""Synchronise les données Pronote vers CalDAV.
:param data: Données Pronote normalisées.
:param settings: Configuration effective de l'exécution.
:return: Résultat de la synchronisation.
:rtype: CalDAVSyncResult
"""
...
def caldav_sync_step(
synchronizer: CalDAVSynchronizer, data: PronoteData, settings: Settings
) -> CalDAVSyncResult:
"""Exécute la synchronisation CalDAV injectée.
:param synchronizer: Service de synchronisation injecté.
:param data: Données Pronote normalisées.
:param settings: Configuration effective de l'exécution.
:return: Résultat CalDAV.
:rtype: CalDAVSyncResult
"""
return synchronizer(data, settings)

View File

@@ -0,0 +1,20 @@
"""Étape de comparaison de l'agenda réel avec l'agenda théorique."""
from __future__ import annotations
from pronote_sync.models.diff import AgendaDiff
from pronote_sync.models.pronote import PronoteData
from pronote_sync.sync.diff import AgendaComparator
def compare_step(comparator: AgendaComparator | None, data: PronoteData) -> AgendaDiff:
"""Compare l'agenda ou retourne un diff vide si la comparaison est désactivée.
:param comparator: Comparateur configuré, ou ``None`` sans agenda théorique.
:param data: Données Pronote normalisées.
:return: Diff d'agenda pour la date cible.
:rtype: AgendaDiff
"""
if comparator is None:
return AgendaDiff(target_date=data.target_date)
return comparator.compare(data.lessons, data.target_date)

View File

@@ -0,0 +1,126 @@
"""Étape de récupération des données Pronote pour une exécution du pipeline."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import date
from pronote_sync.errors import PipelineCriticalError, PipelineWarning
from pronote_sync.models.agenda import Lesson, SchoolEvent
from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message
from pronote_sync.sources.pronote.fallback import PronoteFetcherProtocol
from pronote_sync.utils.redaction import redact_exception
@dataclass(frozen=True)
class FetchedPronoteData:
"""Représente les données brutes récupérées pendant une exécution.
:ivar lessons: Cours récupérés depuis la source sélectionnée.
:ivar homeworks: Devoirs destinés à la date cible.
:ivar school_events: Événements scolaires récupérés avec l'agenda.
:ivar messages: Messages et informations Pronote disponibles.
:ivar target_date: Date cible du digest.
"""
lessons: list[Lesson]
homeworks: list[Homework]
school_events: list[SchoolEvent]
messages: list[Message]
target_date: date
def resolve_target_date(
today: date, lessons: list[Lesson], school_events: list[SchoolEvent]
) -> date:
"""Détermine la date cible du digest à partir de l'agenda disponible.
La règle privilégie J+1 lorsqu'il contient des cours. Si la journée en
cours contient des cours mais pas J+1, le prochain cours connu est choisi.
Sans cours correspondant, J+1 est conservé, y compris pendant les vacances.
:param today: Date de référence de l'exécution.
:param lessons: Cours récupérés pour la fenêtre de synchronisation.
:param school_events: Événements scolaires récupérés (réservés aux évolutions
du libellé de jour sans cours).
:return: Date cible du digest.
:rtype: date
"""
del school_events
tomorrow = date.fromordinal(today.toordinal() + 1)
lesson_dates = {lesson.start.date() for lesson in lessons}
if tomorrow in lesson_dates:
return tomorrow
if today in lesson_dates:
future_dates = sorted(day for day in lesson_dates if day > today)
if future_dates:
return future_dates[0]
return tomorrow
def _fetch_optional_messages(
fetcher: PronoteFetcherProtocol,
) -> tuple[list[Message], list[PipelineWarning]]:
"""Récupère les messages et informations sans bloquer le pipeline.
:param fetcher: Fetcher Pronote configuré.
:return: Messages disponibles et avertissements éventuels.
:rtype: tuple[list[Message], list[PipelineWarning]]
"""
messages: list[Message] = []
warnings: list[PipelineWarning] = []
for step, method in (
("fetch_messages", fetcher.fetch_messages),
("fetch_informations", fetcher.fetch_informations),
):
try:
messages.extend(method())
except Exception as exc:
warnings.append(
PipelineWarning(
f"Récupération non critique échouée : {redact_exception(exc)}",
step=step,
)
)
return messages, warnings
def fetch_step(
fetcher: PronoteFetcherProtocol, *, today: date | None = None
) -> tuple[FetchedPronoteData, list[PipelineWarning]]:
"""Récupère les données Pronote critiques et les compléments dégradables.
L'agenda et les devoirs sont critiques : leur échec empêche de produire un
digest fiable et est donc propagé comme :class:`PipelineCriticalError`.
Les messages et informations sont facultatifs ; leur échec produit un
avertissement et une liste partielle reste valide.
:param fetcher: Fetcher Pronote configuré.
:param today: Date de référence, injectée par les tests ; J courant par défaut.
:return: Données récupérées et avertissements non critiques.
:rtype: tuple[FetchedPronoteData, list[PipelineWarning]]
:raises PipelineCriticalError: Si l'agenda ou les devoirs ne sont pas disponibles.
"""
try:
lessons, school_events = fetcher.fetch_agenda()
target_date = resolve_target_date(today or date.today(), lessons, school_events)
homeworks = fetcher.fetch_homework(target_date)
except PipelineCriticalError:
raise
except Exception as exc:
raise PipelineCriticalError(
f"Récupération Pronote impossible : {redact_exception(exc)}", step="fetch"
) from None
messages, warnings = _fetch_optional_messages(fetcher)
return (
FetchedPronoteData(
lessons=lessons,
homeworks=homeworks,
school_events=school_events,
messages=messages,
target_date=target_date,
),
warnings,
)

View File

@@ -0,0 +1,34 @@
"""Étape de récupération non bloquante des articles RSS du collège."""
from __future__ import annotations
from pronote_sync.models.blog import BlogArticle
from pronote_sync.sources.blog.rss import BlogRSSClient
from pronote_sync.sources.blog.state import BlogRSSState
from pronote_sync.utils.redaction import redact_exception
def fetch_blog_step(client: BlogRSSClient | None, state: BlogRSSState | None) -> list[BlogArticle]:
"""Récupère les articles RSS nouveaux en conservant l'état du client.
:param client: Client RSS configuré, ou ``None`` lorsque le blog est désactivé.
:param state: État de déduplication et de cache HTTP associé au run.
:return: Nouveaux articles du blog.
:rtype: list[BlogArticle]
:raises RuntimeError: Si la récupération RSS injectée échoue.
"""
if client is None or state is None:
return []
try:
etag, last_modified = state.get_cache_headers()
result = client.fetch_and_parse(
known_guids=state.get_known_guids(), etag=etag, last_modified=last_modified
)
if result.error is not None:
raise RuntimeError(result.error) from None
if not result.not_modified:
state.add_guids(article.id for article in result.articles)
state.update_cache_headers(result.etag, result.last_modified)
return list(result.articles)
except Exception as exc:
raise RuntimeError(f"Récupération du blog échouée : {redact_exception(exc)}") from None

View File

@@ -0,0 +1,31 @@
"""Étape de normalisation et d'ordonnancement déterministe des données Pronote."""
from __future__ import annotations
from datetime import datetime
from pronote_sync.models.pronote import PronoteData
from pronote_sync.pipeline.steps.fetch import FetchedPronoteData
def normalize_step(fetched: FetchedPronoteData, *, generated_at: datetime) -> PronoteData:
"""Construit le contrat ``PronoteData`` dans un ordre déterministe.
:param fetched: Données brutes produites par :func:`fetch_step`.
:param generated_at: Horodatage de l'exécution fourni par l'orchestrateur.
:return: Données Pronote normalisées.
:rtype: PronoteData
"""
return PronoteData(
lessons=sorted(fetched.lessons, key=lambda lesson: (lesson.start, lesson.id)),
homeworks=sorted(
fetched.homeworks, key=lambda homework: (homework.due_on, homework.subject, homework.id)
),
school_events=sorted(
fetched.school_events,
key=lambda event: (event.from_date, event.to_date, event.kind.value, event.label),
),
messages=sorted(fetched.messages, key=lambda message: (message.date, message.id)),
target_date=fetched.target_date,
generated_at=generated_at,
)

View File

@@ -0,0 +1,17 @@
"""Étape d'envoi du digest sur le canal de notification."""
from __future__ import annotations
from pronote_sync.channels.protocol import Channel
from pronote_sync.models.xmpp import XmppMessage
def send_step(channel: Channel, message: XmppMessage) -> bool:
"""Envoie le digest et retourne le statut fourni par le canal.
:param channel: Canal de sortie configuré.
:param message: Digest XMPP à transmettre.
:return: ``True`` si l'envoi a réussi, ``False`` sinon.
:rtype: bool
"""
return channel.send(message)

View File

@@ -0,0 +1,21 @@
"""Étape de génération optionnelle de synthèse IA."""
from __future__ import annotations
from pronote_sync.models.synthesis import SynthesisInput, SynthesisResult
from pronote_sync.synthesis.provider import SynthesisProvider
def synthesis_step(
provider: SynthesisProvider | None, input_data: SynthesisInput
) -> SynthesisResult | None:
"""Génère une synthèse lorsque le fournisseur IA est activé.
:param provider: Fournisseur IA optionnel.
:param input_data: Données à synthétiser.
:return: Synthèse produite, ou ``None`` si le fournisseur est désactivé.
:rtype: SynthesisResult | None
"""
if provider is None:
return None
return provider.generate(input_data)

View File

@@ -29,6 +29,8 @@ class BlogRSSFetchResult(BaseModel):
réponse RSS, si elle est disponible. ``None`` par défaut. réponse RSS, si elle est disponible. ``None`` par défaut.
:param not_modified: Vaut ``True`` si le serveur a répondu avec le :param not_modified: Vaut ``True`` si le serveur a répondu avec le
statut ``304 Not Modified``, ``False`` sinon. statut ``304 Not Modified``, ``False`` sinon.
:param error: Message d'erreur expurgé si la récupération a échoué,
``None`` sinon.
""" """
model_config = ConfigDict(frozen=True) model_config = ConfigDict(frozen=True)
@@ -52,3 +54,7 @@ class BlogRSSFetchResult(BaseModel):
default=False, default=False,
description="Vaut True si le serveur a répondu 304 Not Modified", description="Vaut True si le serveur a répondu 304 Not Modified",
) )
error: str | None = Field(
default=None,
description=("Message d'erreur expurgé si la récupération a échoué, None sinon"),
)

View File

@@ -131,12 +131,14 @@ class BlogRSSClient:
if getattr(feed, "bozo", None): if getattr(feed, "bozo", None):
bozo_exception = getattr(feed, "bozo_exception", None) bozo_exception = getattr(feed, "bozo_exception", None)
if bozo_exception is not None: if bozo_exception is not None:
error_msg = f"Flux RSS invalide : {redact_exception(bozo_exception)}"
logger.warning( logger.warning(
"Flux RSS du blog invalide (%s), ignoré : %s", "Flux RSS du blog invalide (%s), ignoré : %s",
redact_exception(bozo_exception), redact_exception(bozo_exception),
redact_url(self.rss_url), redact_url(self.rss_url),
) )
else: else:
error_msg = "Flux RSS invalide"
logger.warning( logger.warning(
"Flux RSS du blog invalide, ignoré : %s", "Flux RSS du blog invalide, ignoré : %s",
redact_url(self.rss_url), redact_url(self.rss_url),
@@ -146,6 +148,7 @@ class BlogRSSClient:
etag=etag, etag=etag,
last_modified=last_modified, last_modified=last_modified,
not_modified=False, not_modified=False,
error=error_msg,
) )
articles: list[BlogArticle] = [] articles: list[BlogArticle] = []
@@ -234,16 +237,18 @@ class BlogRSSClient:
not_modified=False, not_modified=False,
) )
except Exception as exc: except Exception as exc:
error_msg = redact_exception(exc)
logger.error( logger.error(
"Échec de la récupération du flux RSS du blog %s : %s", "Échec de la récupération du flux RSS du blog %s : %s",
redact_url(self.rss_url), redact_url(self.rss_url),
redact_exception(exc), error_msg,
) )
return BlogRSSFetchResult( return BlogRSSFetchResult(
articles=(), articles=(),
etag=etag, etag=etag,
last_modified=last_modified, last_modified=last_modified,
not_modified=False, not_modified=False,
error=error_msg,
) )
@staticmethod @staticmethod

View File

@@ -155,16 +155,17 @@ class PronoteClient:
"""Crée et connecte le client ``pronotepy`` (connexion paresseuse). """Crée et connecte le client ``pronotepy`` (connexion paresseuse).
Le client est créé une seule fois puis réutilisé pour les appels Le client est créé une seule fois puis réutilisé pour les appels
suivants. Le nom d'ENT est résolu via :func:`_resolve_ent` et le suivants. Le nom d'ENT, s'il est configuré, est résolu via
type de compte (``student`` ou ``parent``) détermine la classe de :func:`_resolve_ent` ; en l'absence d'ENT, ``ent=None`` est transmis
client utilisée. L'erreur de connexion est relancée sans à ``pronotepy`` pour une connexion directe. Le type de compte
journalisation, la méthode publique appelante étant responsable (``student`` ou ``parent``) détermine la classe de client utilisée.
de la journaliser. L'erreur de connexion est relancée sans journalisation, la méthode
publique appelante étant responsable de la journaliser.
:return: Le client ``pronotepy`` connecté. :return: Le client ``pronotepy`` connecté.
:rtype: pronotepy.Client :rtype: pronotepy.Client
:raises ValueError: Si ``pronote_url``, ``username``, ``password`` :raises ValueError: Si ``pronote_url``, ``username`` ou ``password``
ou ``ent`` est manquant, ou si l'ENT est inconnu. est manquant, ou si l'ENT fourni est inconnu.
:raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue. :raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue.
""" """
if self._client is None: if self._client is None:
@@ -172,11 +173,9 @@ class PronoteClient:
username = self._settings.username username = self._settings.username
password = self._settings.password password = self._settings.password
ent = self._settings.ent ent = self._settings.ent
if pronote_url is None or username is None or password is None or ent is None: if pronote_url is None or username is None or password is None:
raise ValueError( raise ValueError("pronote_url, username et password sont requis pour pronotepy")
"pronote_url, username, password et ent sont requis pour pronotepy" resolver = _resolve_ent(ent) if ent is not None else None
)
resolver = _resolve_ent(ent)
client_class: type[pronotepy.Client] = ( client_class: type[pronotepy.Client] = (
pronotepy.ParentClient pronotepy.ParentClient
if self._settings.account_type == "parent" if self._settings.account_type == "parent"

View File

@@ -16,6 +16,8 @@ d'origine ne sont jamais chaînées (``from None``).
from __future__ import annotations from __future__ import annotations
import logging import logging
from collections.abc import Iterator
from contextlib import contextmanager
from datetime import date, timedelta from datetime import date, timedelta
from enum import StrEnum from enum import StrEnum
from typing import Literal, Protocol from typing import Literal, Protocol
@@ -100,6 +102,30 @@ class PronoteFetcher:
""" """
self._settings: Settings = settings self._settings: Settings = settings
self._pronote_client: PronoteClientProtocol = pronote_client self._pronote_client: PronoteClientProtocol = pronote_client
self._run_ical_agenda: tuple[list[Lesson], list[SchoolEvent]] | None = None
self._cache_ical_for_run = False
@contextmanager
def run_context(self) -> Iterator[None]:
"""Active un cache iCal éphémère pour une exécution du pipeline.
Le cache couvre à la fois le téléchargement et le parsing du flux.
Il est toujours supprimé à la sortie du contexte, y compris si une
étape échoue : il ne peut donc pas devenir un cache global ou
persistant entre deux exécutions.
:yield: Aucun objet.
:rtype: Iterator[None]
"""
previous_cache = self._run_ical_agenda
previous_enabled = self._cache_ical_for_run
self._run_ical_agenda = None
self._cache_ical_for_run = True
try:
yield
finally:
self._run_ical_agenda = previous_cache
self._cache_ical_for_run = previous_enabled
def _fetch_window(self) -> tuple[date, date]: def _fetch_window(self) -> tuple[date, date]:
"""Calcule la fenêtre de synchronisation autour de la date du jour. """Calcule la fenêtre de synchronisation autour de la date du jour.
@@ -123,8 +149,8 @@ class PronoteFetcher:
def _is_pronotepy_configured(self) -> bool: def _is_pronotepy_configured(self) -> bool:
"""Vérifie que la source pronotepy est entièrement configurée. """Vérifie que la source pronotepy est entièrement configurée.
:return: ``True`` si ``pronote_url``, ``username``, ``password`` :return: ``True`` si ``pronote_url``, ``username`` et ``password``
et ``ent`` sont tous définis, ``False`` sinon. sont tous définis, ``False`` sinon.
:rtype: bool :rtype: bool
""" """
pronote = self._settings.pronote pronote = self._settings.pronote
@@ -132,7 +158,6 @@ class PronoteFetcher:
pronote.pronote_url is not None pronote.pronote_url is not None
and pronote.username is not None and pronote.username is not None
and pronote.password is not None and pronote.password is not None
and pronote.ent is not None
) )
def _fetch_agenda_ical(self) -> tuple[list[Lesson], list[SchoolEvent]]: def _fetch_agenda_ical(self) -> tuple[list[Lesson], list[SchoolEvent]]:
@@ -144,12 +169,17 @@ class PronoteFetcher:
:raises OSError: Si le fichier iCal local est illisible. :raises OSError: Si le fichier iCal local est illisible.
:raises requests.RequestException: Si la récupération HTTP échoue. :raises requests.RequestException: Si la récupération HTTP échoue.
""" """
if self._cache_ical_for_run and self._run_ical_agenda is not None:
return self._run_ical_agenda
ical_url = self._settings.pronote.ical_url ical_url = self._settings.pronote.ical_url
if ical_url is None: if ical_url is None:
raise ValueError("PRONOTE_ICAL_URL est requis pour la source iCal") raise ValueError("PRONOTE_ICAL_URL est requis pour la source iCal")
raw_ical = fetch_ical(ical_url.get_secret_value()) raw_ical = fetch_ical(ical_url.get_secret_value())
lessons, _, school_events = parse_ical(raw_ical) lessons, _, school_events = parse_ical(raw_ical)
return lessons, school_events result = (lessons, school_events)
if self._cache_ical_for_run:
self._run_ical_agenda = result
return result
def _fetch_agenda_pronotepy(self) -> tuple[list[Lesson], list[SchoolEvent]]: def _fetch_agenda_pronotepy(self) -> tuple[list[Lesson], list[SchoolEvent]]:
"""Récupère l'agenda depuis pronotepy. """Récupère l'agenda depuis pronotepy.

View File

@@ -89,7 +89,9 @@ def redact_secrets(text: str, extra_secrets: Iterable[SecretStr | str] = ()) ->
(clés API brutes, jetons, mots de passe, etc.) sont ensuite remplacées (clés API brutes, jetons, mots de passe, etc.) sont ensuite remplacées
littéralement, par ``str.replace``, par ``REDACTED`` dans le texte, y littéralement, par ``str.replace``, par ``REDACTED`` dans le texte, y
compris lorsqu'elles n'apparaissent pas sous une forme ``cle=valeur`` compris lorsqu'elles n'apparaissent pas sous une forme ``cle=valeur``
reconnue. Une valeur vide ou ``None`` est ignorée. reconnue. Une valeur vide ou ``None`` est ignorée. Les secrets sont
appliqués du plus long au plus court afin qu'un secret qui est une
sous-chaîne d'un autre soit remplacé en premier, sans être corrompu.
:param text: Texte pouvant contenir des URLs ou des secrets en clair. :param text: Texte pouvant contenir des URLs ou des secrets en clair.
:param extra_secrets: Itérable de secrets bruts (``str`` ou :param extra_secrets: Itérable de secrets bruts (``str`` ou
@@ -101,19 +103,25 @@ def redact_secrets(text: str, extra_secrets: Iterable[SecretStr | str] = ()) ->
redacted = _URL_PATTERN.sub(lambda match: redact_url(match.group(0)), text) redacted = _URL_PATTERN.sub(lambda match: redact_url(match.group(0)), text)
redacted = _AUTH_HEADER_PATTERN.sub(r"\1: REDACTED", redacted) redacted = _AUTH_HEADER_PATTERN.sub(r"\1: REDACTED", redacted)
redacted = _ISOLATED_SECRET_PATTERN.sub(r"\1\2\3REDACTED", redacted) redacted = _ISOLATED_SECRET_PATTERN.sub(r"\1\2\3REDACTED", redacted)
values: list[str] = []
for secret in extra_secrets: for secret in extra_secrets:
value: str | None = secret.get_secret_value() if isinstance(secret, SecretStr) else secret value: str | None = secret.get_secret_value() if isinstance(secret, SecretStr) else secret
if not value: if not value:
continue continue
values.append(value)
for value in sorted(values, key=len, reverse=True):
redacted = redacted.replace(value, _REDACTED) redacted = redacted.replace(value, _REDACTED)
return redacted return redacted
def redact_exception(exc: Exception) -> str: def redact_exception(exc: Exception, extra_secrets: Iterable[SecretStr | str] = ()) -> str:
"""Masque les secrets dans la représentation textuelle d'une exception. """Masque les secrets dans la représentation textuelle d'une exception.
:param exc: Exception dont le message doit être rédigé. :param exc: Exception dont le message doit être rédigé.
:param extra_secrets: Itérable de secrets bruts (``str`` ou
:class:`pydantic.SecretStr`) à masquer, transmis à
:func:`redact_secrets`. Les valeurs vides ou ``None`` sont ignorées.
:return: Représentation textuelle de l'exception avec les secrets masqués. :return: Représentation textuelle de l'exception avec les secrets masqués.
:rtype: str :rtype: str
""" """
return redact_secrets(str(exc)) return redact_secrets(str(exc), extra_secrets)

View File

@@ -94,7 +94,7 @@ skips = ["B101"] # Ignorer les assertions (utilisées dans les tests)
line-length = 100 line-length = 100
target-version = "py313" target-version = "py313"
# Exclure la documentation markdown (ruff format ne doit pas toucher aux blocs de code Python inclus) # Exclure la documentation markdown (ruff format ne doit pas toucher aux blocs de code Python inclus)
extend-exclude = ["GUIDE_DEV_PYTHON.md"] extend-exclude = ["GUIDE_DEV_PYTHON.md", ".worktrees"]
[tool.ruff.lint] [tool.ruff.lint]
select = [ select = [

245
scripts/check_secrets.py Normal file
View File

@@ -0,0 +1,245 @@
#!/usr/bin/env python3
"""Vérifie l'absence de secrets littéraux avant un déploiement.
Le script inspecte le contenu textuel du dépôt, ou uniquement les fichiers
ajoutés/modifiés dans l'index avec ``--staged``. Il ne transmet jamais la
valeur détectée : les résultats ne contiennent que le chemin, le numéro de
ligne et le type de motif. Les fichiers d'environnement et les répertoires
générés sont exclus, car ils ne doivent pas être versionnés ni déployés.
"""
from __future__ import annotations
import argparse
import re
import subprocess # nosec B404
from collections.abc import Callable, Iterable, Sequence
from dataclasses import dataclass
from pathlib import Path
_EXCLUDED_PARTS = frozenset({".git", ".venv", ".worktrees", "__pycache__", ".."})
_EXCLUDED_NAMES = frozenset({".env", ".secrets.baseline", "GUIDE_DEV_PYTHON.md"})
_EXCLUDED_TOP_LEVEL = frozenset({"tests"})
_ALLOWLIST_MARKER = "secret-check: allow"
_UNQUOTED_CONFIG_SUFFIXES = frozenset({".conf", ".ini", ".toml", ".yaml", ".yml"})
_TEXT_SUFFIXES = frozenset(
{".conf", ".ini", ".json", ".md", ".py", ".service", ".timer", ".toml", ".txt", ".yaml", ".yml"}
)
_LITERAL_SECRET_RE = re.compile(
r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)"
r"\s*[:=]\s*['\"][^'\"\r\n]{3,}['\"]"
)
_UNQUOTED_SECRET_RE = re.compile(
r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)"
r"\s*[:=]\s*[a-z0-9][a-z0-9._~+/-]{2,}"
)
_URL_SECRET_RE = re.compile(
r"(?ix)[?&](?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)"
r"=([^&#\s]{3,})"
)
_EXTRA_NAMES = frozenset({"pronote_sync"})
@dataclass(frozen=True)
class SecretFinding:
"""Représente un motif sensible détecté sans exposer sa valeur.
:ivar path: Chemin relatif du fichier concerné.
:ivar line: Numéro de ligne du motif.
:ivar rule: Règle ayant détecté le motif.
"""
path: Path
line: int
rule: str
CommandRunner = Callable[..., subprocess.CompletedProcess[str]]
#: Fournisseur de contenu pour un chemin relatif ; retourne ``None`` pour ignorer.
ContentProvider = Callable[[Path], str | None]
def _is_candidate(path: Path) -> bool:
"""Indique si un chemin peut être analysé comme fichier texte.
Les fichiers de déploiement sans extension, nommés explicitement dans
``_EXTRA_NAMES``, sont également retenus.
:param path: Chemin relatif au dépôt.
:return: ``True`` lorsque le fichier est textuel et non exclu.
:rtype: bool
"""
return (
not path.is_absolute()
and path.name not in _EXCLUDED_NAMES
and path.parts[0] not in _EXCLUDED_TOP_LEVEL
and not any(part in _EXCLUDED_PARTS for part in path.parts)
and (path.suffix in _TEXT_SUFFIXES or path.name in _EXTRA_NAMES)
)
def _repository_files(root: Path) -> list[Path]:
"""Liste les fichiers textuels présents dans le dépôt de travail.
Les tests et la spécification historique ne font pas partie de l'artefact
déployé : leurs sentinelles et exemples intentionnels ne doivent donc pas
bloquer le déploiement.
:param root: Racine du dépôt à analyser.
:return: Chemins relatifs triés des fichiers analysables.
:rtype: list[Path]
"""
return sorted(
path.relative_to(root)
for path in root.rglob("*")
if path.is_file() and _is_candidate(path.relative_to(root))
)
def _staged_files(root: Path, runner: CommandRunner) -> list[Path]:
"""Retourne les fichiers ajoutés ou modifiés actuellement indexés.
:param root: Racine du dépôt Git.
:param runner: Exécuteur de sous-processus injectable pour les tests.
:return: Chemins relatifs triés des fichiers indexés analysables.
:rtype: list[Path]
:raises RuntimeError: Si Git ne peut pas fournir les fichiers indexés.
"""
result = runner(
["git", "diff", "--cached", "--name-only", "-z", "--diff-filter=ACMR"],
cwd=root,
capture_output=True,
text=True,
check=False,
)
if result.returncode != 0:
raise RuntimeError("Impossible de lister les fichiers Git indexés") from None
paths = [Path(value) for value in result.stdout.split("\0") if value]
return sorted(path for path in paths if _is_candidate(path))
def _staged_content_provider(root: Path, runner: CommandRunner) -> ContentProvider:
"""Retourne un lecteur de contenu depuis l'index Git.
Lit le blob indexé via ``git show :<chemin>`` afin de ne pas dépendre de
l'état du working tree, dont la copie de travail peut différer de l'index.
:param root: Racine du dépôt Git.
:param runner: Exécuteur de sous-processus injectable pour les tests.
:return: Fonction de lecture du contenu indexé ; ``None`` si indisponible.
:rtype: ContentProvider
"""
def provider(relative_path: Path) -> str | None:
result = runner(
["git", "show", f":{relative_path}"],
cwd=root,
capture_output=True,
text=True,
check=False,
)
if result.returncode != 0:
return None
return result.stdout
return provider
def find_secrets(
root: Path,
files: Iterable[Path],
content_provider: ContentProvider | None = None,
) -> list[SecretFinding]:
"""Détecte les motifs de secrets littéraux dans les fichiers désignés.
Les lignes explicitement marquées ``secret-check: allow`` sont exclues :
cette échappatoire doit rester locale à une fixture ou un exemple contrôlé.
:param root: Racine du dépôt analysé.
:param files: Chemins relatifs à inspecter.
:param content_provider: Lecteur optionnel du contenu d'un fichier ; par
défaut le contenu est lu depuis le working tree via ``read_text``.
Si le lecteur retourne ``None`` ou lève une erreur d'encodage, le
fichier est ignoré.
:return: Résultats triés par chemin, ligne et règle.
:rtype: list[SecretFinding]
"""
findings: list[SecretFinding] = []
for relative_path in files:
path = root / relative_path
try:
if content_provider is not None:
content = content_provider(relative_path)
else:
content = path.read_text(encoding="utf-8")
if content is None:
continue
except (OSError, UnicodeDecodeError):
continue
for number, line in enumerate(content.splitlines(), start=1):
if _ALLOWLIST_MARKER in line:
continue
is_literal_secret = _LITERAL_SECRET_RE.search(line) or (
relative_path.suffix in _UNQUOTED_CONFIG_SUFFIXES
and _UNQUOTED_SECRET_RE.search(line)
)
if is_literal_secret:
findings.append(SecretFinding(relative_path, number, "affectation-litterale"))
if _URL_SECRET_RE.search(line):
findings.append(SecretFinding(relative_path, number, "parametre-url"))
return sorted(findings, key=lambda finding: (str(finding.path), finding.line, finding.rule))
def _parse_arguments(arguments: Sequence[str] | None = None) -> argparse.Namespace:
"""Analyse les options de vérification.
:param arguments: Arguments explicites, ou ``None`` pour ceux du processus.
:return: Options validées.
:rtype: argparse.Namespace
"""
parser = argparse.ArgumentParser(description="Vérifie les secrets avant déploiement.")
parser.add_argument(
"--staged",
action="store_true",
help="Analyse uniquement les fichiers ajoutés ou modifiés dans l'index Git.",
)
return parser.parse_args(arguments)
def main(
arguments: Sequence[str] | None = None,
*,
root: Path | None = None,
runner: CommandRunner = subprocess.run,
) -> int:
"""Exécute la vérification de secrets et retourne un code de sortie.
:param arguments: Arguments de ligne de commande.
:param root: Racine à analyser ; le dépôt du script par défaut.
:param runner: Exécuteur Git injectable pour les tests.
:return: ``0`` sans motif, ``1`` si un motif est trouvé, ``2`` si le contrôle échoue.
:rtype: int
"""
parsed_arguments = _parse_arguments(arguments)
repository_root = root or Path(__file__).resolve().parents[1]
try:
if parsed_arguments.staged:
files = _staged_files(repository_root, runner)
content_provider = _staged_content_provider(repository_root, runner)
else:
files = _repository_files(repository_root)
content_provider = None
except RuntimeError as error:
print(f"ERREUR: {error}")
return 2
findings = find_secrets(repository_root, files, content_provider=content_provider)
if not findings:
print("OK: aucun secret littéral détecté.")
return 0
for finding in findings:
print(f"ECHEC: {finding.path}:{finding.line} ({finding.rule})")
return 1
if __name__ == "__main__":
raise SystemExit(main())

View File

@@ -14,6 +14,7 @@ from pydantic import SecretStr
from pronote_sync.config.settings import CalDAVSettings from pronote_sync.config.settings import CalDAVSettings
from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent, SchoolEventKind from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent, SchoolEventKind
from pronote_sync.models.homework import Homework from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message, MessageType
from pronote_sync.models.pronote import PronoteData from pronote_sync.models.pronote import PronoteData
@@ -85,19 +86,38 @@ def caldav_settings() -> CalDAVSettings:
) )
@pytest.fixture
def sample_message() -> Message:
"""Message Pronote pour les tests.
:return: Message Pronote de test.
:rtype: Message
"""
return Message(
id="msg-001",
type=MessageType.INFORMATION,
title="Information de rentrée",
content="La rentrée est prévue le 1er septembre.",
author="Administration",
date=datetime(2026, 1, 15, 9, 0),
read=False,
)
@pytest.fixture @pytest.fixture
def pronote_data( def pronote_data(
sample_lesson: Lesson, sample_lesson: Lesson,
sample_cancelled_lesson: Lesson, sample_cancelled_lesson: Lesson,
sample_homework: Homework, sample_homework: Homework,
sample_school_event: SchoolEvent, sample_school_event: SchoolEvent,
sample_message: Message,
) -> PronoteData: ) -> PronoteData:
"""Données Pronote de test avec des cours, devoirs et événements.""" """Données Pronote de test avec des cours, devoirs, événements et messages."""
return PronoteData( return PronoteData(
lessons=[sample_lesson, sample_cancelled_lesson], lessons=[sample_lesson, sample_cancelled_lesson],
homeworks=[sample_homework], homeworks=[sample_homework],
school_events=[sample_school_event], school_events=[sample_school_event],
messages=[], messages=[sample_message],
target_date=date(2026, 1, 15), target_date=date(2026, 1, 15),
generated_at=datetime(2026, 1, 15, 0, 0), generated_at=datetime(2026, 1, 15, 0, 0),
) )

1
tests/e2e/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Tests end-to-end de l'interface en ligne de commande."""

189
tests/e2e/test_cli.py Normal file
View 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

56
tests/fixtures/pronote-6e.ics vendored Normal file
View File

@@ -0,0 +1,56 @@
BEGIN:VCALENDAR
VERSION:2.0
PRODID:-//Index Education//Pronote//FR
X-WR-CALNAME:Classe de 6e
BEGIN:VEVENT
UID:Edt_22222@index-education.net-20260908T140000Z-Index-Education
DTSTAMP:20260908T140000Z
DTSTART:20260908T140000Z
DTEND:20260908T150000Z
SUMMARY:SVT
CATEGORIES:Cours
DESCRIPTION:<div>
Matière : SVT
Professeur : M. Dubois
Salle : 104
Groupe : Classe entière
<strong>Contenu pédagogique :
</strong>
Découverte de la cellule et de ses constituants.
<strong>Pour le 15/09/2026 :
</strong>
Lire le chapitre 2 et schématiser une cellule végétale.
<strong>Donné le 08/09/2026 :
</strong>
Lire le chapitre 2 et schématiser une cellule végétale.
</div>
END:VEVENT
BEGIN:VEVENT
UID:Edt_33333@index-education.net-20260908T140000Z-Index-Education
DTSTAMP:20260908T140000Z
DTSTART:20260909T100000Z
DTEND:20260909T110000Z
SUMMARY:Histoire-Géographie
CATEGORIES:Cours - Cours modifié
DESCRIPTION:<div>
Matière : Histoire-Géographie
Professeur : Mme Lefevre
Salle : 203
Groupe : Classe entière
<strong>Contenu pédagogique :
</strong>
Les grands repères du temps long : la Préhistoire.
</div>
END:VEVENT
BEGIN:VEVENT
UID:Edt_44444@index-education.net-20260908T140000Z-Index-Education
DTSTAMP:20260908T140000Z
DTSTART;VALUE=DATE:20260928
DTEND;VALUE=DATE:20260929
SUMMARY:Sortie pédagogique
CATEGORIES:Sortie scolaire
DESCRIPTION:Journée de sortie pédagogique au musée d'histoire naturelle.
END:VEVENT
END:VCALENDAR

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,249 @@
"""Tests unitaires du contrôle de secrets de déploiement."""
from __future__ import annotations
import importlib.util
import subprocess
import sys
from pathlib import Path
from types import ModuleType
from typing import TYPE_CHECKING
import pytest
if TYPE_CHECKING:
from _pytest.capture import CaptureFixture
@pytest.fixture
def secret_checker() -> ModuleType:
"""Charge le script de vérification sans l'exécuter comme programme.
:return: Module du script de contrôle de secrets.
:rtype: ModuleType
"""
script_path = Path(__file__).parents[2] / "scripts" / "check_secrets.py"
specification = importlib.util.spec_from_file_location("check_secrets", script_path)
assert specification is not None
assert specification.loader is not None
module = importlib.util.module_from_spec(specification)
sys.modules[specification.name] = module
try:
specification.loader.exec_module(module)
finally:
del sys.modules[specification.name]
return module
def test_main_accepts_clean_files_and_ignores_environment_file(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
"""Vérifie qu'un dépôt propre réussit sans analyser le fichier d'environnement.
:param secret_checker: Module du script sous test.
:param tmp_path: Répertoire temporaire représentant un dépôt.
:param capsys: Fixture de capture de sortie.
:return: None
"""
(tmp_path / "application.py").write_text("value = 'safe'\n", encoding="utf-8")
ignored_environment_secret = 'password = "private-value"\n' # pragma: allowlist secret
(tmp_path / ".env").write_text(
ignored_environment_secret, encoding="utf-8"
) # secret-check: allow
assert secret_checker.main([], root=tmp_path) == 0
assert "OK:" in capsys.readouterr().out
def test_main_reports_a_literal_secret_without_disclosing_its_value(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
"""Vérifie qu'un secret littéral échoue sans fuite de sa valeur.
:param secret_checker: Module du script sous test.
:param tmp_path: Répertoire temporaire représentant un dépôt.
:param capsys: Fixture de capture de sortie.
:return: None
"""
sentinel = "m14-literal-sentinel"
(tmp_path / "settings.py").write_text(
f'password = "{sentinel}"\n', encoding="utf-8"
) # secret-check: allow
assert secret_checker.main([], root=tmp_path) == 1
output = capsys.readouterr().out
assert "settings.py:1 (affectation-litterale)" in output
assert sentinel not in output
def test_main_reports_an_unquoted_configuration_secret_without_disclosing_its_value(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
"""Vérifie qu'un secret de configuration non cité échoue sans fuite de sa valeur.
:param secret_checker: Module du script sous test.
:param tmp_path: Répertoire temporaire représentant un dépôt.
:param capsys: Fixture de capture de sortie.
:return: None
"""
sentinel = "m14-unquoted-sentinel"
(tmp_path / "settings.yaml").write_text(
f"password: {sentinel}\n", encoding="utf-8"
) # secret-check: allow
assert secret_checker.main([], root=tmp_path) == 1
output = capsys.readouterr().out
assert "settings.yaml:1 (affectation-litterale)" in output
assert sentinel not in output
def test_main_detects_sensitive_url_parameter(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
"""Vérifie qu'un paramètre URL sensible déclenche un échec.
:param secret_checker: Module du script sous test.
:param tmp_path: Répertoire temporaire représentant un dépôt.
:param capsys: Fixture de capture de sortie.
:return: None
"""
sentinel = "m14-url-sentinel"
(tmp_path / "settings.yaml").write_text(
f"url: https://example.invalid/calendar?icalsecurise={sentinel}\n", encoding="utf-8"
) # secret-check: allow
assert secret_checker.main([], root=tmp_path) == 1
output = capsys.readouterr().out
assert "settings.yaml:1 (parametre-url)" in output
assert sentinel not in output
def test_staged_mode_inspects_only_paths_provided_by_git(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
"""Vérifie que l'option staged ignore les fichiers non indexés.
:param secret_checker: Module du script sous test.
:param tmp_path: Répertoire temporaire représentant un dépôt.
:param capsys: Fixture de capture de sortie.
:return: None
"""
(tmp_path / "indexed.py").write_text("answer = 42\n", encoding="utf-8")
untracked_secret = 'api_key = "m14-untracked-sentinel"\n' # pragma: allowlist secret
(tmp_path / "untracked.py").write_text(
untracked_secret, encoding="utf-8"
) # secret-check: allow
def runner(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
"""Simule Git avec un seul fichier indexé.
:return: Résultat Git simulé.
:rtype: subprocess.CompletedProcess[str]
"""
return subprocess.CompletedProcess([], 0, stdout="indexed.py\0", stderr="")
assert secret_checker.main(["--staged"], root=tmp_path, runner=runner) == 0
assert "OK:" in capsys.readouterr().out
def test_main_detects_prefixed_secret_assignment(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
"""Vérifie qu'une variable préfixée (PRONOTE_PASSWORD) est détectée.
:param secret_checker: Module du script sous test.
:param tmp_path: Répertoire temporaire représentant un dépôt.
:param capsys: Fixture de capture de sortie.
:return: None
"""
sentinel = "m14-prefixed-secret"
(tmp_path / "config.py").write_text(
f'PRONOTE_PASSWORD = "{sentinel}"\n', encoding="utf-8"
) # secret-check: allow
assert secret_checker.main([], root=tmp_path) == 1
output = capsys.readouterr().out
assert "config.py:1" in output
assert sentinel not in output
def test_main_detects_short_secret_assignment(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
"""Vérifie qu'un secret court (< 8 caractères) est détecté.
:param secret_checker: Module du script sous test.
:param tmp_path: Répertoire temporaire représentant un dépôt.
:param capsys: Fixture de capture de sortie.
:return: None
"""
sentinel = "s3cr3t"
(tmp_path / "config.py").write_text(
f'password = "{sentinel}"\n', encoding="utf-8"
) # secret-check: allow
assert secret_checker.main([], root=tmp_path) == 1
output = capsys.readouterr().out
assert "config.py:1" in output
assert sentinel not in output
def test_staged_mode_reads_index_content_not_working_tree(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
"""Vérifie que --staged lit le contenu indexé, pas le working tree.
:param secret_checker: Module du script sous test.
:param tmp_path: Répertoire temporaire représentant un dépôt.
:param capsys: Fixture de capture de sortie.
:return: None
"""
indexed_secret = "m14-indexed-only-secret" # pragma: allowlist secret
(tmp_path / "staged.py").write_text(
f'password = "{indexed_secret}"\n', encoding="utf-8"
) # secret-check: allow
(tmp_path / "staged.py").write_text('value = "safe"\n', encoding="utf-8")
def runner(*args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
"""Simule Git en renvoyant le contenu indexé pour le blob demandé.
:return: Résultat Git simulé.
:rtype: subprocess.CompletedProcess[str]
"""
first_argument = args[0] if args else []
command = (
[str(argument) for argument in first_argument]
if isinstance(first_argument, list)
else []
)
if "show" in command:
return subprocess.CompletedProcess(
command, 0, stdout=f'password = "{indexed_secret}"\n', stderr=""
)
return subprocess.CompletedProcess(command, 0, stdout="staged.py\0", stderr="")
assert secret_checker.main(["--staged"], root=tmp_path, runner=runner) == 1
output = capsys.readouterr().out
assert "staged.py:1" in output
assert indexed_secret not in output
def test_main_scans_extensionless_deployment_file(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
"""Vérifie qu'un fichier de déploiement sans extension est scanné.
:param secret_checker: Module du script sous test.
:param tmp_path: Répertoire temporaire représentant un dépôt.
:param capsys: Fixture de capture de sortie.
:return: None
"""
sentinel = "m14-logrotate-secret"
(tmp_path / "pronote_sync").write_text(
f'password = "{sentinel}"\n', encoding="utf-8"
) # secret-check: allow
assert secret_checker.main([], root=tmp_path) == 1
output = capsys.readouterr().out
assert "pronote_sync:1" in output
assert sentinel not in output

View File

@@ -1088,6 +1088,102 @@ def test_homework_sources_auto_no_source_configured_raises(mock_fetcher: Pronote
assert "ni la source iCal ni pronotepy n'est configurée" in str(exc_info.value) assert "ni la source iCal ni pronotepy n'est configurée" in str(exc_info.value)
def test_fetch_agenda_auto_ical_configured_fails_fallback_to_pronotepy_without_ent(
mock_fetcher: PronoteFetcher,
) -> None:
"""Test le mode auto : iCal configuré mais échoue, repli sur pronotepy sans ent.
On mock iCal pour échouer, pronotepy configuré sans ent. On vérifie que pronotepy est appelé
et que le résultat provient de pronotepy, pas une erreur.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
start_dt = datetime(2025, 9, 1, 8, 0)
end_dt = datetime(2025, 9, 1, 9, 30)
lessons = [
Lesson(
id="l1",
start=start_dt,
end=end_dt,
subject="Maths",
teachers=("Dupont",),
rooms=("S1",),
group="2ndeA",
status=LessonStatus.NORMAL,
content=None,
)
]
# Override settings to use auto mode with ical_url configured but ent=None
mock_fetcher._settings.pronote.agenda_source = "auto"
mock_fetcher._settings.pronote.ent = None # Explicitly None
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
):
m_fetch_ical.side_effect = OSError("iCal unreachable")
m_parse_ical.side_effect = OSError("iCal parse error")
client = MagicMock()
client.get_lessons.return_value = lessons
mock_fetcher._pronote_client = client
result_lessons, result_events = mock_fetcher.fetch_agenda()
assert result_lessons == lessons
assert result_events == []
client.get_lessons.assert_called_once()
def test_fetch_homework_auto_ical_configured_fails_fallback_to_pronotepy_without_ent(
mock_fetcher: PronoteFetcher,
) -> None:
"""Test le mode auto des devoirs : iCal configuré mais échoue, repli sur pronotepy sans ent.
On mock iCal pour échouer, pronotepy configuré sans ent. On vérifie que pronotepy est appelé
et que le résultat provient de pronotepy, pas une erreur.
:param mock_fetcher: Fetcher de test.
:return: None
:rtype: None
"""
target_date = date(2025, 9, 10)
homeworks = [
Homework(
id="hw1",
subject="Physique",
teachers=(),
assigned_on=None,
due_on=target_date,
text="TP à préparer",
html="TP à préparer",
)
]
# Override settings to use auto mode with ical_url configured but ent=None
mock_fetcher._settings.pronote.homework_source = "auto"
mock_fetcher._settings.pronote.ent = None # Explicitly None
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
patch("pronote_sync.sources.pronote.fallback.collect_homeworks") as m_collect,
):
m_fetch_ical.side_effect = OSError("iCal unreachable")
m_parse_ical.side_effect = OSError("iCal parse error")
client = MagicMock()
client.get_homeworks.return_value = homeworks
mock_fetcher._pronote_client = client
m_collect.return_value = homeworks
result = mock_fetcher.fetch_homework(target_date)
assert result == homeworks
client.get_homeworks.assert_called_once()
def test_fetch_homework_fallback_both_fail_raises_pipeline_critical_error( def test_fetch_homework_fallback_both_fail_raises_pipeline_critical_error(
mock_fetcher: PronoteFetcher, mock_fetcher: PronoteFetcher,
) -> None: ) -> None:
@@ -1127,7 +1223,7 @@ def test_fetch_homework_fallback_both_fail_raises_pipeline_critical_error(
def test_fetch_homework_auto_fallback_returns_empty_logs_warning( def test_fetch_homework_auto_fallback_returns_empty_logs_warning(
mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture mock_fetcher: PronoteFetcher, caplog: pytest.LogCaptureFixture
) -> None: ) -> None:
"""Test que fetch_homework retourne [] et journalise un avertissement si le repli retourne vide. """Test que fetch_homework retourne [] et journalise un avertissement si le repli est vide.
On mock ICAL pour échouer, pronotepy configuré et retourne vide. On vérifie le retour et le log. On mock ICAL pour échouer, pronotepy configuré et retourne vide. On vérifie le retour et le log.
Ce test utilise le mode AUTO pour tester le comportement de repli. Ce test utilise le mode AUTO pour tester le comportement de repli.
@@ -1186,4 +1282,99 @@ def test_fetch_informations_logs_and_re_raises_secret(
) )
def test_is_pronotepy_configured_without_ent_returns_true() -> None:
"""Test _is_pronotepy_configured() retourne True quand ent est None.
Les autres champs de la configuration pronotepy sont présents, donc la
fonction renvoie True.
Ce test valide que PRONOTE_ENT est optionnel pour pronotepy.
:return: None
"""
from pronote_sync.config.settings import PronoteSettings, Settings
from pronote_sync.sources.pronote.fallback import PronoteFetcher
# Créer des settings avec pronotepy configuré mais sans ent
settings = Settings(
pronote=PronoteSettings(
pronote_url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent=None, # Explicitement None
agenda_source="pronotepy",
homework_source="pronotepy",
),
app=Settings().app,
)
client: _MockPronoteClientProtocol = MagicMock()
fetcher = PronoteFetcher(settings=settings, pronote_client=client)
# Should return True even without ent
assert fetcher._is_pronotepy_configured() is True
def test_agenda_sources_auto_without_ical_and_without_ent_returns_pronotepy() -> None:
"""Test _agenda_sources() en mode AUTO sans iCal URL et sans ent retourne pronotepy.
Ce test valide que le mode auto peut utiliser pronotepy même sans ent configuré.
:return: None
"""
from pronote_sync.config.settings import PronoteSettings, Settings
from pronote_sync.sources.pronote.fallback import PronoteFetcher
settings = Settings(
pronote=PronoteSettings(
pronote_url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent=None, # Explicitement None
ical_url=None, # Pas de iCal URL
agenda_source="auto",
),
app=Settings().app,
)
client: _MockPronoteClientProtocol = MagicMock()
fetcher = PronoteFetcher(settings=settings, pronote_client=client)
primary, fallback = fetcher._agenda_sources()
assert primary == "pronotepy"
assert fallback is None
def test_homework_sources_auto_without_ical_and_without_ent_returns_pronotepy() -> None:
"""Test _homework_sources() en mode AUTO sans iCal URL et sans ent retourne pronotepy.
Ce test valide que le mode auto peut utiliser pronotepy pour les devoirs même sans ent configuré.
:return: None
"""
from pronote_sync.config.settings import PronoteSettings, Settings
from pronote_sync.sources.pronote.fallback import PronoteFetcher
settings = Settings(
pronote=PronoteSettings(
pronote_url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent=None, # Explicitement None
ical_url=None, # Pas de iCal URL
homework_source="auto",
),
app=Settings().app,
)
client: _MockPronoteClientProtocol = MagicMock()
fetcher = PronoteFetcher(settings=settings, pronote_client=client)
primary, fallback = fetcher._homework_sources()
assert primary == "pronotepy"
assert fallback is None
# Ensure trailing newline # Ensure trailing newline

View File

@@ -309,7 +309,9 @@ def test_missing_credentials_raises(empty_pronote_settings: PronoteSettings) ->
""" """
client = PronoteClient(empty_pronote_settings) client = PronoteClient(empty_pronote_settings)
with pytest.raises(ValueError, match="pronote_url, username, password et ent sont requis"): with pytest.raises(
ValueError, match="pronote_url, username et password sont requis pour pronotepy"
):
client._connect() client._connect()
@@ -365,6 +367,132 @@ def test_connect_parent_account_type(
pronotepy.Client.assert_not_called() # type: ignore[attr-defined] pronotepy.Client.assert_not_called() # type: ignore[attr-defined]
def test_connect_without_ent_but_with_required_credentials(
mocker: pytest_mock.MockerFixture,
) -> None:
"""Vérifie que _connect() fonctionne sans ent mais avec les autres identifiants requis.
Ce test valide que PRONOTE_ENT est optionnel pour une connexion directe Pronote.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
from unittest.mock import Mock
from pronote_sync.config.settings import PronoteSettings
from pronote_sync.sources.pronote.client import PronoteClient
# Settings sans ent mais avec les autres champs requis
settings = PronoteSettings(
pronote_url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent=None, # Explicitement None
account_type="parent",
)
mock_client = mocker.MagicMock()
mock_client_class = Mock(return_value=mock_client)
mocker.patch("pronotepy.ParentClient", new=mock_client_class)
mocker.patch("pronotepy.Client")
client = PronoteClient(settings)
connected_client = client._connect()
# Should not raise ValueError about missing ent
assert connected_client is mock_client
# Verify ParentClient was called with ent=None
mock_client_class.assert_called_once_with(
pronote_url="https://pronote.example.com",
username="testuser",
password="testpass", # pragma: allowlist secret
ent=None, # ent should be None, not resolved
)
def test_connect_missing_required_credentials_still_raises(
mocker: pytest_mock.MockerFixture,
) -> None:
"""Vérifie que _connect() lève ValueError si pronote_url, username ou password manquent.
Ce test valide que l'erreur ne mentionne plus ent comme requis.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
from pronote_sync.config.settings import PronoteSettings
from pronote_sync.sources.pronote.client import PronoteClient
# Settings avec ent mais sans pronote_url
settings = PronoteSettings(
pronote_url=None,
username="testuser",
password=SecretStr("testpass"),
ent=None,
account_type="parent",
)
client = PronoteClient(settings)
with pytest.raises(ValueError) as exc_info:
client._connect()
# Error should NOT mention ent as required
assert "pronote_url, username et password sont requis" in str(exc_info.value)
assert "ent" not in str(exc_info.value)
def test_connect_missing_username_raises(mocker: pytest_mock.MockerFixture) -> None:
"""Vérifie que _connect() lève ValueError si username manque.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
from pronote_sync.config.settings import PronoteSettings
from pronote_sync.sources.pronote.client import PronoteClient
settings = PronoteSettings(
pronote_url="https://pronote.example.com",
username=None,
password=SecretStr("testpass"),
ent=None,
account_type="parent",
)
client = PronoteClient(settings)
with pytest.raises(ValueError) as exc_info:
client._connect()
assert "pronote_url, username et password sont requis" in str(exc_info.value)
def test_connect_missing_password_raises(mocker: pytest_mock.MockerFixture) -> None:
"""Vérifie que _connect() lève ValueError si password manque.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
from pronote_sync.config.settings import PronoteSettings
from pronote_sync.sources.pronote.client import PronoteClient
settings = PronoteSettings(
pronote_url="https://pronote.example.com",
username="testuser",
password=None,
ent=None,
account_type="parent",
)
client = PronoteClient(settings)
with pytest.raises(ValueError) as exc_info:
client._connect()
assert "pronote_url, username et password sont requis" in str(exc_info.value)
def test_connect_student_account_type( def test_connect_student_account_type(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None: ) -> None:
@@ -398,6 +526,57 @@ def test_connect_student_account_type(
pronotepy.ParentClient.assert_not_called() # type: ignore[attr-defined] pronotepy.ParentClient.assert_not_called() # type: ignore[attr-defined]
def test_connect_with_ent_resolution_still_works(
mocker: pytest_mock.MockerFixture,
) -> None:
"""Vérifie que _resolve_ent est appelé et fonctionne quand ent est fourni.
Ce test valide que lorsque ent est fourni, il est toujours résolu via _resolve_ent.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
from unittest.mock import Mock
from pronote_sync.config.settings import PronoteSettings
from pronote_sync.sources.pronote.client import PronoteClient
settings = PronoteSettings(
pronote_url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent="bordeaux", # ent est fourni
account_type="parent",
)
mock_client = mocker.MagicMock()
mock_client_class = Mock(return_value=mock_client)
mocker.patch("pronotepy.ParentClient", new=mock_client_class)
# Mock _resolve_ent to return a mock resolver
mock_resolver = Mock()
mocker.patch(
"pronote_sync.sources.pronote.client._resolve_ent",
return_value=mock_resolver,
)
client = PronoteClient(settings)
_ = client._connect()
# _resolve_ent should have been called
from pronote_sync.sources.pronote.client import _resolve_ent as resolve_ent_func
resolve_ent_func.assert_called_once_with("bordeaux") # type: ignore[attr-defined]
# ParentClient should have been called with the resolved ent
mock_client_class.assert_called_once_with(
pronote_url="https://pronote.example.com",
username="testuser",
password="testpass", # pragma: allowlist secret
ent=mock_resolver,
)
def test_get_messages_degraded_on_error( def test_get_messages_degraded_on_error(
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
) -> None: ) -> None: