Compare commits

..
1 Commits
Author SHA1 Message Date
Codex 5bd97402bf fix(cli): exposer les executions degradees 2026-09-12 15:37:11 +02:00
15 changed files with 174 additions and 1078 deletions
+1 -8
View File
@@ -32,7 +32,7 @@ CALDAV_CALENDAR_PATH=/pronote-sync/
# Autoriser HTTP (non-HTTPS) pour un serveur CalDAV local (localhost uniquement) # Autoriser HTTP (non-HTTPS) pour un serveur CalDAV local (localhost uniquement)
CALDAV_ALLOW_INSECURE_HTTP=false CALDAV_ALLOW_INSECURE_HTTP=false
# Fenêtre de synchronisation (jours) — entier >= 0 ; 0 = aucune journée supplémentaire de ce côté (le jour courant reste inclus). Valeurs négatives refusées au chargement (ValidationError). # Fenêtre de synchronisation (jours)
SYNC_PAST_DAYS=7 SYNC_PAST_DAYS=7
SYNC_FUTURE_DAYS=30 SYNC_FUTURE_DAYS=30
@@ -66,13 +66,6 @@ AI_BASE_URL=https://api.openai.com/v1
# AI_API_KEY= # AI_API_KEY=
# AI_MODEL=gpt-4o-mini # exemple recommandé, non activé par défaut # AI_MODEL=gpt-4o-mini # exemple recommandé, non activé par défaut
# NOTE : la validation structurelle de AI_BASE_URL s'applique à TOUS les
# providers (openai, litellm, openai-compatible) : HTTPS obligatoire sauf si
# AI_ALLOW_INSECURE_HTTP=true, aucun credential embarqué (user:pass@hôte),
# aucun paramètre sensible dans la query string (token, key, api_key,
# secret, password, auth), et aucune manipulation automatique de /v1.
# Seul le provider openai-compatible exige AI_BASE_URL et AI_MODEL.
# Exemple : OpenRouter (HTTPS) # Exemple : OpenRouter (HTTPS)
# AI_PROVIDER=openai-compatible # AI_PROVIDER=openai-compatible
# AI_BASE_URL=https://openrouter.ai/api/v1 # AI_BASE_URL=https://openrouter.ai/api/v1
+2 -2
View File
@@ -140,7 +140,7 @@
"filename": "GUIDE_DEV_PYTHON.md", "filename": "GUIDE_DEV_PYTHON.md",
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa", "hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
"is_verified": false, "is_verified": false,
"line_number": 5186 "line_number": 5125
} }
], ],
"tests/unit/test_caldav_gateway.py": [ "tests/unit/test_caldav_gateway.py": [
@@ -185,5 +185,5 @@
} }
] ]
}, },
"generated_at": "2026-09-12T17:57:39Z" "generated_at": "2026-09-12T12:04:07Z"
} }
+3 -11
View File
@@ -159,12 +159,9 @@ pronote-sync --dry-run
- Après chaque login réussi, les credentials exportées par `pronotepy.export_credentials()` sont - Après chaque login réussi, les credentials exportées par `pronotepy.export_credentials()` sont
persistées dans `.pronote_auth_state.json` (permissions `0600`, format JSON versionné, écriture persistées dans `.pronote_auth_state.json` (permissions `0600`, format JSON versionné, écriture
atomique). Le token rotate à chaque session et peut également être rafraîchi pendant l'exécution atomique). Le token rotate à chaque session et peut également être rafraîchi pendant l'exécution
(refresh automatique pronotepy après une `PronoteAPIError`). La persistance s'applique après (refresh automatique pronotepy après une `PronoteAPIError`). Les credentials sont persistées après
chaque login réussi **et après chaque opération de données réussie** (agenda, devoirs, messages) chaque login réussi **et après chaque opération de données réussie** (agenda, devoirs, messages,
pour garantir la persistance du token valide ; seules les opérations qui se connectent réellement informations) pour garantir la persistance du token valide.
et récupèrent des données déclenchent la persistance. En mode `qr_token`, `get_informations()`
est ignorée (retour immédiat `[]` sans connexion ni verrou) et ne déclenche donc aucune
persistance.
- Les logins suivants utilisent `pronotepy.token_login(**credentials)` avec le token persisté. - Les logins suivants utilisent `pronotepy.token_login(**credentials)` avec le token persisté.
- En cas d'échec de `token_login` (token expiré/invalide), une `PronoteAuthRotationError` est levée. - En cas d'échec de `token_login` (token expiré/invalide), une `PronoteAuthRotationError` est levée.
Cette erreur se propage sans wrapping à travers `PronoteFetcher` et `fetch_step` jusqu'à Cette erreur se propage sans wrapping à travers `PronoteFetcher` et `fetch_step` jusqu'à
@@ -175,11 +172,6 @@ pronote-sync --dry-run
- `PronoteAuthRotationError` est re-levée telle quelle (`except PronoteAuthRotationError: raise`) - `PronoteAuthRotationError` est re-levée telle quelle (`except PronoteAuthRotationError: raise`)
dans toutes les couches d'enveloppement du chemin critique (fetch_agenda, fetch_homework, dans toutes les couches d'enveloppement du chemin critique (fetch_agenda, fetch_homework,
fetch_step). Ne pas l'attraper avec `except Exception` sans la re-léver d'abord. fetch_step). Ne pas l'attraper avec `except Exception` sans la re-léver d'abord.
- En mode `qr_token`, `get_informations()` retourne **inconditionnellement** une liste vide
(`[]`) sans connexion, verrou, chargement d'état ni appel réseau, et journalise un message
INFO unique : l'endpoint `PageActualites` renvoie une erreur pronotepy 20 sur les instances
HubEduConnect testées, provoquant un refresh redondant du token. Ce contournement n'est pas
configurable ; aucun état anti-répétition n'est conservé.
- Le fichier `.pronote_auth_state.json` ne doit jamais être committé (couvert par `.gitignore`). - Le fichier `.pronote_auth_state.json` ne doit jamais être committé (couvert par `.gitignore`).
Son contenu (token vivant) ne doit jamais apparaître dans les logs, les messages d'erreur ou Son contenu (token vivant) ne doit jamais apparaître dans les logs, les messages d'erreur ou
les notifications XMPP. les notifications XMPP.
+5 -1
View File
@@ -25,13 +25,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- `PRONOTE_URL` ignoré à cause du double préfixe `env_prefix` (renommage `pronote_url``url` dans `PronoteSettings`) - `PRONOTE_URL` ignoré à cause du double préfixe `env_prefix` (renommage `pronote_url``url` dans `PronoteSettings`)
- `PRONOTE_ENT` rendu optionnel pour les connexions pronotepy directes - `PRONOTE_ENT` rendu optionnel pour les connexions pronotepy directes
- `.env.example` corrigé (`eleve.html``parent.html`) - `.env.example` corrigé (`eleve.html``parent.html`)
- #20/#21`get_informations()` ignorée en mode `qr_token` : retourne `[]` immédiatement, sans connexion, verrou ni appel réseau, évitant l'échec systématique de l'endpoint `PageActualites` (erreur pronotepy 20 sur les instances HubEduConnect testées) et le refresh redondant du token associé.
### Changed ### Changed
- Wiki `GuidePronote` enrichi : section "Quand l'ENT est obligatoire" (EduConnect/HubEduConnect), exemple Bordeaux - Wiki `GuidePronote` enrichi : section "Quand l'ENT est obligatoire" (EduConnect/HubEduConnect), exemple Bordeaux
- `AGENTS.md` : ajout de la section §13 "Versionnage et releases" - `AGENTS.md` : ajout de la section §13 "Versionnage et releases"
### Known Issues
- #20 — Triple authentification pronotepy (double INIT + refresh) lors d'un run
- #21 — Erreur pronotepy 20 « La page a expiré ! (11) » sur `get_informations`
### Tests ### Tests
- 694 tests passés, couverture 94.93% - 694 tests passés, couverture 94.93%
+46 -107
View File
@@ -299,18 +299,18 @@ d'un besoin réel et testé.
| `PRONOTE_AGENDA_SOURCE` | Source pour l'agenda (`auto`, `ical`, `pronotepy`). | `auto` | `Literal` | | `PRONOTE_AGENDA_SOURCE` | Source pour l'agenda (`auto`, `ical`, `pronotepy`). | `auto` | `Literal` |
| `PRONOTE_HOMEWORK_SOURCE` | Source pour les devoirs (`auto`, `ical`, `pronotepy`). | `auto` | `Literal` | | `PRONOTE_HOMEWORK_SOURCE` | Source pour les devoirs (`auto`, `ical`, `pronotepy`). | `auto` | `Literal` |
| `PRONOTE_MESSAGES_SOURCE` | Source pour les messages (`pronotepy` uniquement). | `pronotepy` | `Literal` | | `PRONOTE_MESSAGES_SOURCE` | Source pour les messages (`pronotepy` uniquement). | `pronotepy` | `Literal` |
| `SYNC_PAST_DAYS` | Nombre de jours dans le passé pour la sync CalDAV, entier `>= 0` (`0` = aucune journée supplémentaire de ce côté ; le jour courant reste inclus). | `7` | `int` | | `SYNC_PAST_DAYS` | Nombre de jours dans le passé pour la sync CalDAV. | `7` | `int` |
| `SYNC_FUTURE_DAYS` | Nombre de jours dans le futur pour la sync CalDAV, entier `>= 0` (`0` = aucune journée supplémentaire de ce côté ; le jour courant reste inclus). | `30` | `int` | | `SYNC_FUTURE_DAYS` | Nombre de jours dans le futur pour la sync CalDAV. | `30` | `int` |
| `THEORETICAL_AGENDA_PATH` | Chemin vers le fichier JSON de l'agenda théorique. | `None` | `str \| None`| | `THEORETICAL_AGENDA_PATH` | Chemin vers le fichier JSON de l'agenda théorique. | `None` | `str \| None`|
| `SCHOOL_HOLIDAYS_PATH` | Chemin vers le fichier JSON des vacances scolaires. | `None` | `str \| None`| | `SCHOOL_HOLIDAYS_PATH` | Chemin vers le fichier JSON des vacances scolaires. | `None` | `str \| None`|
| `THEORETICAL_WEEK_ANCHOR_DATE` | Date de référence pour la parité des semaines (paire/impaire). | `None` | `date \| None`| | `THEORETICAL_WEEK_ANCHOR_DATE` | Date de référence pour la parité des semaines (paire/impaire). | `None` | `date \| None`|
| `THEORETICAL_WEEK_ANCHOR_TYPE` | Parité de la semaine de référence (`even` ou `odd`). | `None` | `Literal["even", "odd"] \| None`| | `THEORETICAL_WEEK_ANCHOR_TYPE` | Parité de la semaine de référence (`even` ou `odd`). | `None` | `Literal["even", "odd"] \| None`|
| `AI_ENABLED` | Activer la synthèse IA. | `False` | `bool` | | `AI_ENABLED` | Activer la synthèse IA. | `False` | `bool` |
| `AI_PROVIDER` | Fournisseur IA (`openai`, `openai-compatible` ou `litellm`). | `openai` | `Literal["openai", "litellm", "openai-compatible"]` | | `AI_PROVIDER` | Fournisseur IA (`openai`, `openai-compatible` ou `litellm`). | `openai` | `Literal["openai", "litellm", "openai-compatible"]` |
| `AI_BASE_URL` | URL de base pour l'API IA (validée structurellement pour tous les providers, voir ci-dessous). | `None` | `str \| None`| | `AI_BASE_URL` | URL de base pour l'API IA (ex: OpenAI compatible). | `None` | `str \| None`|
| `AI_API_KEY` | Clé API pour l'API IA. | `None` | `SecretStr` | | `AI_API_KEY` | Clé API pour l'API IA. | `None` | `SecretStr` |
| `AI_MODEL` | Modèle IA à utiliser (exemple recommandé : `gpt-4o-mini`). | `None` | `str \| None`| | `AI_MODEL` | Modèle IA à utiliser (exemple recommandé : `gpt-4o-mini`). | `None` | `str \| None`|
| `AI_ALLOW_INSECURE_HTTP` | Autoriser HTTP (non sécurisé) pour tous les providers (openai, litellm, openai-compatible). | `False` | `bool` | | `AI_ALLOW_INSECURE_HTTP` | Autoriser HTTP (non sécurisé) pour `openai-compatible` uniquement. | `False` | `bool` |
| `DRY_RUN` | Simulation sans sortie distante ni état local persistant ; incompatible avec `qr_token`. | `False` | `bool` | | `DRY_RUN` | Simulation sans sortie distante ni état local persistant ; incompatible avec `qr_token`. | `False` | `bool` |
| `LOG_LEVEL` | Niveau de log (`DEBUG`, `INFO`, `WARNING`, `ERROR`). | `INFO` | `str` | | `LOG_LEVEL` | Niveau de log (`DEBUG`, `INFO`, `WARNING`, `ERROR`). | `INFO` | `str` |
@@ -394,7 +394,7 @@ LOG_LEVEL=INFO
> `CalDAVSettings.calendar_path` a pour valeur par défaut `"/pronote-sync/"`. > `CalDAVSettings.calendar_path` a pour valeur par défaut `"/pronote-sync/"`.
> `XmppSettings.resource` a pour valeur par défaut `"pronote-sync"`. > `XmppSettings.resource` a pour valeur par défaut `"pronote-sync"`.
> > ``AISettings.provider`` accepte également ``openai-compatible`` (réutilise ``OpenAISynthesisProvider`` avec un ``base_url`` personnalisé). > > ``AISettings.provider`` accepte également ``openai-compatible`` (réutilise ``OpenAISynthesisProvider`` avec un ``base_url`` personnalisé).
> > ``AISettings.allow_insecure_http`` (défaut ``False``) autorise les URLs HTTP pour les trois providers (``openai``, ``litellm`` et ``openai-compatible``). > > ``AISettings.allow_insecure_http`` (défaut ``False``) autorise les URLs HTTP pour le provider ``openai-compatible`` uniquement.
```python ```python
from typing import Literal from typing import Literal
@@ -421,8 +421,8 @@ class CalDAVSettings(BaseSettings):
password: SecretStr | None = None password: SecretStr | None = None
calendar_path: str = "/pronote-sync/" calendar_path: str = "/pronote-sync/"
allow_insecure_http: bool = False allow_insecure_http: bool = False
sync_past_days: int = Field(default=7, ge=0) sync_past_days: int = 7
sync_future_days: int = Field(default=30, ge=0) sync_future_days: int = 30
class AISettings(BaseSettings): class AISettings(BaseSettings):
@@ -2279,12 +2279,6 @@ méthodes agenda/devoirs ne transforment jamais une erreur en liste vide : elles
version expurgée puis lèvent une erreur expurgée avec `from None`. Les méthodes de messages et version expurgée puis lèvent une erreur expurgée avec `from None`. Les méthodes de messages et
d'informations sont non critiques et peuvent retourner une liste vide avec un warning. d'informations sont non critiques et peuvent retourner une liste vide avec un warning.
En mode `qr_token`, `get_informations()` est ignorée : elle retourne immédiatement `[]` sans
connexion, verrou ni appel réseau, et journalise un message INFO unique. L'endpoint
`PageActualites` renvoie en effet une erreur pronotepy 20 sur les instances HubEduConnect
testées, provoquant un refresh redondant du token. Ce comportement n'est pas configurable ;
`get_messages()` n'est pas concernée par ce contournement.
Les objets renvoyés par `client.homework(start, end)` couvrent une fenêtre. Le résultat destiné à Les objets renvoyés par `client.homework(start, end)` couvrent une fenêtre. Le résultat destiné à
un jour cible est donc filtré explicitement sur `homework.date == target_date`. un jour cible est donc filtré explicitement sur `homework.date == target_date`.
@@ -2296,13 +2290,11 @@ d'authentification et de récupération par un verrou POSIX local non bloquant,
`.pronote_auth_state.json.lock`, à côté de `.pronote_auth_state.json`. `.pronote_auth_state.json.lock`, à côté de `.pronote_auth_state.json`.
Le verrou couvre l'ensemble du cycle QR/token : chargement de l'état, connexion par token ou Le verrou couvre l'ensemble du cycle QR/token : chargement de l'état, connexion par token ou
enrôlement QR initial, opération de données (agenda, devoirs, messages ; informations hors mode enrôlement QR initial, opération de données (agenda, devoirs, messages ou informations), puis
`qr_token`), puis persistance des credentials actualisées. En mode `qr_token`, `get_informations()` persistance des credentials actualisées. Une tentative concurrente échoue immédiatement avec une
est ignorée (retour immédiat `[]` sans connexion ni verrou) : elle n'acquiert pas le verrou et ne erreur d'état d'authentification expurgée ; elle ne patiente pas et ne relance pas
déclenche aucune persistance. Une tentative concurrente échoue immédiatement avec une erreur d'état l'authentification. Le contenu du token, le PIN et les autres credentials ne sont jamais inclus
d'authentification expurgée ; elle ne patiente pas et ne relance pas l'authentification. Le contenu dans les logs ni dans ce message d'erreur.
du token, le PIN et les autres credentials ne sont jamais inclus dans les logs ni dans ce message
d'erreur.
Ce mécanisme est un contrat **local** : il coordonne des processus sur le même hôte Linux et un Ce mécanisme est un contrat **local** : il coordonne des processus sur le même hôte Linux et un
filesystem local. Pour des déploiements conteneurisés, les conteneurs qui partagent le même compte filesystem local. Pour des déploiements conteneurisés, les conteneurs qui partagent le même compte
@@ -3925,26 +3917,16 @@ L'import de `litellm` est conditionnel avec `try/except ImportError` → `None`.
| ``openai-compatible`` | Proxy ou serveur compatible OpenAI | ``OpenAISynthesisProvider`` | | ``openai-compatible`` | Proxy ou serveur compatible OpenAI | ``OpenAISynthesisProvider`` |
| ``litellm`` | Bibliothèque LiteLLM embarquée | ``LiteLLMSynthesisProvider`` | | ``litellm`` | Bibliothèque LiteLLM embarquée | ``LiteLLMSynthesisProvider`` |
Pour le provider ``openai-compatible``, ``AI_BASE_URL`` et ``AI_MODEL`` sont requis ; pour Pour le provider ``openai-compatible``, la validation de la configuration est stricte :
``openai`` et ``litellm``, ils sont optionnels. La validation structurelle de ``AI_BASE_URL``
(partagée via ``_validate_base_url``) s'applique de façon identique aux trois providers dès que
l'URL est renseignée :
- URL parsable par ``urlparse`` (``ValueError`` ou schéma vide → refusée) et hostname non vide. - ``AI_BASE_URL`` est requis.
- Schéma limité à ``http``/``https`` ; ``http`` refusé sauf si ``AI_ALLOW_INSECURE_HTTP=true``. - ``AI_MODEL`` est requis et ne doit pas être vide.
- Credentials dans l'URL (``user:pass@host``) refusés. - ``AI_API_KEY`` est requis (MVP).
- Paramètres sensibles dans la *query string* refusés, y compris sans valeur - L'URL doit utiliser le schéma ``https`` sauf si ``AI_ALLOW_INSECURE_HTTP=true``.
(``token``, ``key``, ``api_key``, ``secret``, ``password``, ``auth``). - Les credentials dans l'URL sont refusés.
- Les paramètres sensibles dans la *query string* sont refusés.
- Aucune manipulation automatique de ``/v1`` n'est effectuée. - Aucune manipulation automatique de ``/v1`` n'est effectuée.
- En cas de violation, un avertissement est journalisé (URL masquée via ``redact_url``) et la - Si la configuration est incomplète, la factory retourne ``None`` avec un avertissement (mode dégradé).
factory retourne ``None`` (mode dégradé) ; la factory ne lève jamais d'exception et ne fait
aucun appel réseau.
| Provider | `AI_BASE_URL` | `AI_MODEL` | Validation structurelle |
|---------------------|---------------|------------|---------------------------------------------|
| ``openai`` | optionnel | optionnel | `_validate_base_url("openai", ...)` |
| ``litellm`` | optionnel | optionnel | `_validate_base_url("litellm", ...)` |
| ``openai-compatible`` | requis | requis | `_validate_openai_compatible_config` (présence puis `_validate_base_url`) |
La politique hors réseau de la table des modèles litellm est gérée par `LITELLM_LOCAL_MODEL_COST_MAP=true`. Les tests utilisent `pytest.importorskip("litellm")`. La politique hors réseau de la table des modèles litellm est gérée par `LITELLM_LOCAL_MODEL_COST_MAP=true`. Les tests utilisent `pytest.importorskip("litellm")`.
@@ -3960,69 +3942,37 @@ from ..utils.redaction import redact_url
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_SENSITIVE_QUERY_PARAMS = {"token", "key", "api_key", "secret", "password", "auth"}
def _validate_base_url(provider: str, url: str, allow_insecure_http: bool) -> str | None:
"""Valide structurellement une URL de base IA, partagée entre providers."""
try:
parsed = urlparse(url)
if not parsed.scheme:
logger.warning("URL invalide pour le provider %s : %s", provider, redact_url(url))
return None
if not parsed.hostname:
logger.warning(
"URL sans hostname pour le provider %s : %s", provider, redact_url(url)
)
return None
if parsed.scheme not in ("http", "https"):
logger.warning(
"Schéma d'URL non supporté pour le provider %s : %s", provider, redact_url(url)
)
return None
if parsed.scheme == "http" and not allow_insecure_http:
logger.warning(
"URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true pour le provider %s : %s",
provider,
redact_url(url),
)
return None
if parsed.username is not None or parsed.password is not None:
logger.warning(
"Credentials dans l'URL refusés pour le provider %s : %s",
provider,
redact_url(url),
)
return None
param_names = [
name.lower() for name, _ in parse_qsl(parsed.query, keep_blank_values=True)
]
if any(name in _SENSITIVE_QUERY_PARAMS for name in param_names):
logger.warning(
"Paramètres sensibles dans l'URL refusés pour le provider %s : %s",
provider,
redact_url(url),
)
return None
# Accéder à parsed.port peut lever ValueError (port invalide/hors bornes).
parsed.port # noqa: B018
except ValueError:
logger.warning("URL invalide pour le provider %s : %s", provider, redact_url(url))
return None
return url
def _validate_openai_compatible_config( def _validate_openai_compatible_config(
url: str | None, model: str | None, allow_insecure_http: bool url: str | None, model: str | None, allow_insecure_http: bool
) -> str | None: ) -> str | None:
"""Valide la configuration du provider ``openai-compatible``.""" """Valide la configuration du provider ``openai-compatible``."""
if not url: if not url or not model:
logger.warning("URL de base requise pour le provider openai-compatible")
return None return None
if not model: try:
logger.warning("Modèle requis pour le provider openai-compatible") parsed = urlparse(url)
except ValueError:
logger.warning("URL invalide : %s", redact_url(url))
return None return None
return _validate_base_url("openai-compatible", url, allow_insecure_http) if not parsed.hostname:
logger.warning("URL sans hostname : %s", redact_url(url))
return None
if parsed.scheme not in ("http", "https"):
return None
if parsed.scheme == "http" and not allow_insecure_http:
return None
if parsed.username is not None or parsed.password is not None:
logger.warning("Credentials dans l'URL refusés : %s", redact_url(url))
return None
sensitive_names = {"token", "key", "api_key", "secret", "password", "auth"}
param_names = [
name.lower() for name, _ in parse_qsl(parsed.query, keep_blank_values=True)
]
if any(name in sensitive_names for name in param_names):
logger.warning(
"Paramètres sensibles dans l'URL refusés : %s", redact_url(url)
)
return None
return url
def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None: def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
@@ -4031,10 +3981,7 @@ def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
Retourne ``None`` lorsque la synthèse IA est désactivée ou qu'aucune clé Retourne ``None`` lorsque la synthèse IA est désactivée ou qu'aucune clé
API n'est configurée. Pour le provider ``litellm``, le paquet ``litellm`` API n'est configurée. Pour le provider ``litellm``, le paquet ``litellm``
(extra ``ai-litellm``) est requis : s'il est absent, un avertissement est (extra ``ai-litellm``) est requis : s'il est absent, un avertissement est
journalisé et ``None`` est retourné. Pour ``openai`` et ``litellm``, journalisé et ``None`` est retourné.
une ``base_url`` éventuelle est validée par :func:`_validate_base_url` ;
pour le provider ``openai-compatible``, la configuration est validée par
:func:`_validate_openai_compatible_config`.
:param settings: Paramètres IA. :param settings: Paramètres IA.
:return: Le fournisseur configuré, ou ``None`` si désactivé ou sans clé API. :return: Le fournisseur configuré, ou ``None`` si désactivé ou sans clé API.
@@ -4054,10 +4001,6 @@ def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
except ImportError: except ImportError:
logger.warning("Extra 'ai-litellm' requis pour le provider litellm") logger.warning("Extra 'ai-litellm' requis pour le provider litellm")
return None return None
if base_url is not None and (
_validate_base_url("litellm", base_url, settings.allow_insecure_http) is None
):
return None
return LiteLLMSynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model) return LiteLLMSynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model)
if settings.provider == "openai-compatible": if settings.provider == "openai-compatible":
@@ -4068,10 +4011,6 @@ def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
return None return None
return OpenAISynthesisProvider(api_key=settings.api_key, base_url=url, model=model) return OpenAISynthesisProvider(api_key=settings.api_key, base_url=url, model=model)
if base_url is not None and (
_validate_base_url("openai", base_url, settings.allow_insecure_http) is None
):
return None
return OpenAISynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model) return OpenAISynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model)
``` ```
+1 -1
View File
@@ -75,7 +75,7 @@ The following variables can be safely pre-configured in `/etc/pronote-sync/prono
- `XMPP_USE_TLS` is deprecated but still supported (aliased to `XMPP_TLS_MODE`) - `XMPP_USE_TLS` is deprecated but still supported (aliased to `XMPP_TLS_MODE`)
- **AI:** - **AI:**
- `AI_ENABLED`, `AI_PROVIDER`, `AI_BASE_URL`, `AI_MODEL`, `AI_ALLOW_INSECURE_HTTP` — la validation structurelle de `AI_BASE_URL` s'applique à tous les providers (HTTPS sauf `AI_ALLOW_INSECURE_HTTP=true`, pas de credentials ni de paramètres sensibles dans l'URL, pas de `/v1` automatique) ; seul `openai-compatible` exige `AI_BASE_URL` et `AI_MODEL`. - `AI_ENABLED`, `AI_PROVIDER`, `AI_BASE_URL`, `AI_MODEL`, `AI_ALLOW_INSECURE_HTTP`
- **Blog:** - **Blog:**
- `BLOG_ENABLED`, `BLOG_RSS_URL` - `BLOG_ENABLED`, `BLOG_RSS_URL`
+11 -4
View File
@@ -105,10 +105,17 @@ sudo systemctl start pronote-sync.service
sudo systemctl status pronote-sync.service sudo systemctl status pronote-sync.service
``` ```
Une exécution en échec laisse l'unité `pronote-sync.service` en état `failed`. La CLI expose un contrat de sortie stable : `0` signifie une exécution complète,
La supervision de l'hôte doit donc déclencher une alerte sur cet état ou sur un `2` une exécution dégradée (les données Pronote sont disponibles mais une étape
échec du timer/service ; le transport de cette alerte (courriel, XMPP ou système optionnelle, CalDAV ou XMPP a échoué), et `1` un échec critique. Tout code non
de supervision) relève de l'exploitation locale. nul laisse l'unité `pronote-sync.service` en état `failed` ; la supervision doit
donc alerter sur cet état ou sur le code de sortie. Le code `2` permet de
distinguer automatiquement une alerte dégradée d'une panne critique, sans lire
les journaux.
Le `--dry-run` n'écrit ni dans CalDAV/XMPP ni dans l'état local. Il conserve le
même contrat de codes : `0` si la simulation est complète, `2` si elle est
dégradée et `1` si elle est critique.
## Journaux et alertes ## Journaux et alertes
+23 -4
View File
@@ -11,6 +11,7 @@ from pydantic import SecretStr
from pronote_sync.config.env import load_settings from pronote_sync.config.env import load_settings
from pronote_sync.config.settings import Settings from pronote_sync.config.settings import Settings
from pronote_sync.errors import ErrorSeverity, PipelineError
from pronote_sync.pipeline.run import PipelineRunner from pronote_sync.pipeline.run import PipelineRunner
from pronote_sync.utils.logging import setup_logging from pronote_sync.utils.logging import setup_logging
from pronote_sync.utils.redaction import redact_secrets from pronote_sync.utils.redaction import redact_secrets
@@ -19,6 +20,26 @@ logger = logging.getLogger(__name__)
_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") _LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")
# Contrat stable pour systemd et les outils de supervision.
EXIT_SUCCESS = 0
EXIT_CRITICAL = 1
EXIT_DEGRADED = 2
def _pipeline_exit_code(data: object | None, errors: Sequence[PipelineError]) -> int:
"""Convertit le résultat du pipeline en code de sortie supervisable.
:param data: Données normalisées produites, ou ``None`` en cas d'échec critique.
:param errors: Erreurs et avertissements de l'exécution.
:return: ``0`` si complet, ``2`` si dégradé, ``1`` si critique.
:rtype: int
"""
if data is None or any(error.severity == ErrorSeverity.CRITICAL for error in errors):
return EXIT_CRITICAL
if errors:
return EXIT_DEGRADED
return EXIT_SUCCESS
def _parse_arguments(arguments: Sequence[str] | None = None) -> argparse.Namespace: def _parse_arguments(arguments: Sequence[str] | None = None) -> argparse.Namespace:
"""Analyse les options de lancement du programme. """Analyse les options de lancement du programme.
@@ -120,7 +141,7 @@ def main(arguments: Sequence[str] | None = None) -> int:
bruts afin de préserver le diagnostic sans exposer de secret. bruts afin de préserver le diagnostic sans exposer de secret.
:param arguments: Arguments optionnels, principalement utiles aux appels programmatiques. :param arguments: Arguments optionnels, principalement utiles aux appels programmatiques.
:return: ``0`` en cas de succès, ``1`` sinon (après analyse des arguments). :return: Code machine-readable : ``0`` complet, ``2`` dégradé, ``1`` critique.
:rtype: int :rtype: int
:raises SystemExit: Si argparse rejette les arguments (code de sortie 2). :raises SystemExit: Si argparse rejette les arguments (code de sortie 2).
""" """
@@ -147,9 +168,7 @@ def main(arguments: Sequence[str] | None = None) -> int:
secrets = _settings_secrets(settings) secrets = _settings_secrets(settings)
for error in errors: for error in errors:
logger.error("%s", redact_secrets(error.message, extra_secrets=secrets)) logger.error("%s", redact_secrets(error.message, extra_secrets=secrets))
if data is None: return _pipeline_exit_code(data, errors)
return 1
return 0
if __name__ == "__main__": if __name__ == "__main__":
+2 -2
View File
@@ -293,8 +293,8 @@ class AppSettings(BaseSettings):
school_holidays_path: str | None = None school_holidays_path: str | None = None
theoretical_week_anchor_date: date | None = None theoretical_week_anchor_date: date | None = None
theoretical_week_anchor_type: Literal["even", "odd"] | None = None theoretical_week_anchor_type: Literal["even", "odd"] | None = None
sync_past_days: int = Field(default=7, ge=0) sync_past_days: int = 7
sync_future_days: int = Field(default=30, ge=0) sync_future_days: int = 30
class Settings(BaseSettings): class Settings(BaseSettings):
+1 -15
View File
@@ -488,23 +488,9 @@ class PronoteClient:
Chaque entrée est mappée sur un modèle :class:`Message` de type Chaque entrée est mappée sur un modèle :class:`Message` de type
``SURVEY`` si c'est un sondage, ``INFORMATION`` sinon. ``SURVEY`` si c'est un sondage, ``INFORMATION`` sinon.
En mode ``qr_token``, la récupération est ignorée sans connexion ni :return: Liste des informations et sondages ; vide en cas d'erreur.
appel réseau : l'endpoint ``PageActualites`` renvoie une erreur
pronotepy 20 sur les instances HubEduConnect testées, provoquant un
refresh redondant du token. La méthode retourne alors immédiatement
une liste vide et journalise un message INFO unique ; ce comportement
n'est pas configurable.
:return: Liste des informations et sondages ; vide en cas d'erreur ou
en mode ``qr_token``.
:rtype: list[Message] :rtype: list[Message]
""" """
if self._settings.auth_mode == "qr_token":
logger.info(
"Récupération des informations Pronote ignorée : endpoint "
"PageActualites indisponible en mode d'authentification qr_token."
)
return []
with self._qr_token_operation_lock(): with self._qr_token_operation_lock():
try: try:
client = self._connect() client = self._connect()
+50 -87
View File
@@ -15,83 +15,20 @@ logger = logging.getLogger(__name__)
__all__ = ["get_synthesis_provider", "SynthesisProvider", "OpenAISynthesisProvider"] __all__ = ["get_synthesis_provider", "SynthesisProvider", "OpenAISynthesisProvider"]
_SENSITIVE_QUERY_PARAMS = {"token", "key", "api_key", "secret", "password", "auth"}
def _validate_base_url(provider: str, url: str, allow_insecure_http: bool) -> str | None:
"""Valide structurellement une URL de base IA, partagée entre providers.
Applique les règles structurelles identiques aux trois providers
(``openai``, ``litellm`` et ``openai-compatible``) : URL parsable par
``urlparse`` (``ValueError`` rejeté), hostname non vide, schéma limité
à ``http``/``https`` (HTTP refusé sauf si ``allow_insecure_http`` vaut
``True``), absence d'identifiants dans le netloc et de paramètres
sensibles dans la requête (y compris les paramètres sans valeur). L'URL
est retournée strictement inchangée : aucune manipulation automatique
du suffixe ``/v1`` n'est effectuée. En cas de violation, un
avertissement est journalisé (l'URL est toujours masquée via
:func:`redact_url`) et ``None`` est retourné ; la fonction ne lève
jamais d'exception et n'effectue aucun appel réseau.
:param provider: Nom du provider (utilisé pour le message d'avertissement).
:param url: URL de base à valider (non vide).
:param allow_insecure_http: Autorise ou non les URLs en HTTP.
:return: L'URL validée, strictement inchangée, ou ``None`` si invalide.
:rtype: str | None
"""
try:
parsed = urlparse(url)
if not parsed.scheme:
logger.warning("URL invalide pour le provider %s : %s", provider, redact_url(url))
return None
if not parsed.hostname:
logger.warning("URL sans hostname pour le provider %s : %s", provider, redact_url(url))
return None
if parsed.scheme not in ("http", "https"):
logger.warning(
"Schéma d'URL non supporté pour le provider %s : %s", provider, redact_url(url)
)
return None
if parsed.scheme == "http" and not allow_insecure_http:
logger.warning(
"URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true pour le provider %s : %s",
provider,
redact_url(url),
)
return None
if parsed.username is not None or parsed.password is not None:
logger.warning(
"Credentials dans l'URL refusés pour le provider %s : %s",
provider,
redact_url(url),
)
return None
param_names = [name.lower() for name, _ in parse_qsl(parsed.query, keep_blank_values=True)]
if any(name in _SENSITIVE_QUERY_PARAMS for name in param_names):
logger.warning(
"Paramètres sensibles dans l'URL refusés pour le provider %s : %s",
provider,
redact_url(url),
)
return None
# Accéder à parsed.port peut lever ValueError (port invalide/hors bornes).
parsed.port # noqa: B018
except ValueError:
logger.warning("URL invalide pour le provider %s : %s", provider, redact_url(url))
return None
return url
def _validate_openai_compatible_config( def _validate_openai_compatible_config(
url: str | None, model: str | None, allow_insecure_http: bool url: str | None, model: str | None, allow_insecure_http: bool
) -> str | None: ) -> str | None:
"""Valide la configuration du provider ``openai-compatible``. """Valide la configuration du provider ``openai-compatible``.
Vérifie d'abord la présence de l'URL de base et du modèle (spécifique Vérifie la présence de l'URL de base et du modèle, le schéma de l'URL
à ``openai-compatible``), puis délègue les règles structurelles (HTTPS obligatoire, HTTP accepté uniquement si ``allow_insecure_http``
partagées à :func:`_validate_base_url`. En cas d'échec, un vaut ``True``), la présence d'un hostname non vide, l'absence
avertissement est journalisé et ``None`` est retourné : la synthèse IA d'identifiants dans le netloc et de paramètres sensibles dans la
se dégrade silencieusement, sans jamais lever d'exception. requête (y compris les paramètres sans valeur). Une URL malformée
(``ValueError`` levé par ``urlparse``) est également rejetée. En cas
d'échec, un avertissement est journalisé (l'URL est toujours masquée
via :func:`redact_url`) et ``None`` est retourné : la synthèse IA se
dégrade silencieusement, sans jamais lever d'exception.
:param url: URL de base de l'API compatible OpenAI. :param url: URL de base de l'API compatible OpenAI.
:param model: Identifiant du modèle à utiliser. :param model: Identifiant du modèle à utiliser.
@@ -106,7 +43,42 @@ def _validate_openai_compatible_config(
if not model: if not model:
logger.warning("Modèle requis pour le provider openai-compatible") logger.warning("Modèle requis pour le provider openai-compatible")
return None return None
return _validate_base_url("openai-compatible", url, allow_insecure_http)
try:
parsed = urlparse(url)
except ValueError:
logger.warning(
"URL invalide pour le provider openai-compatible : %s",
redact_url(url),
)
return None
if not parsed.hostname:
logger.warning(
"URL sans hostname pour le provider openai-compatible : %s",
redact_url(url),
)
return None
if parsed.scheme not in ("http", "https"):
logger.warning(
"Schéma d'URL non supporté pour le provider openai-compatible : %s",
redact_url(url),
)
return None
if parsed.scheme == "http" and not allow_insecure_http:
logger.warning(
"URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true : %s",
redact_url(url),
)
return None
if parsed.username is not None or parsed.password is not None:
logger.warning("Credentials dans l'URL refusés : %s", redact_url(url))
return None
sensitive_names = {"token", "key", "api_key", "secret", "password", "auth"}
param_names = [name.lower() for name, _ in parse_qsl(parsed.query, keep_blank_values=True)]
if any(name in sensitive_names for name in param_names):
logger.warning("Paramètres sensibles dans l'URL refusés : %s", redact_url(url))
return None
return url
def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None: def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
@@ -115,15 +87,14 @@ def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
Retourne ``None`` lorsque la synthèse IA est désactivée ou qu'aucune clé Retourne ``None`` lorsque la synthèse IA est désactivée ou qu'aucune clé
API n'est configurée. Pour le provider ``litellm``, le paquet ``litellm`` API n'est configurée. Pour le provider ``litellm``, le paquet ``litellm``
(extra ``ai-litellm``) est requis : s'il est absent, un avertissement est (extra ``ai-litellm``) est requis : s'il est absent, un avertissement est
journalisé et ``None`` est retourné. Pour ``openai`` et ``litellm``, journalisé et ``None`` est retourné. Pour le provider
une ``base_url`` éventuelle est validée par :func:`_validate_base_url` ; ``openai-compatible``, la configuration (URL de base et modèle) est
pour le provider ``openai-compatible``, la configuration (URL de base validée par :func:`_validate_openai_compatible_config` ; en cas de
et modèle) est validée par :func:`_validate_openai_compatible_config` ; rejet, ``None`` est retourné avec un avertissement.
en cas de rejet, ``None`` est retourné avec un avertissement.
:param settings: Paramètres IA. :param settings: Paramètres IA.
:return: Le fournisseur configuré, ou ``None`` si désactivé, sans clé API :return: Le fournisseur configuré, ou ``None`` si désactivé, sans clé API
ou avec une configuration invalide. ou avec une configuration ``openai-compatible`` invalide.
:rtype: SynthesisProvider | None :rtype: SynthesisProvider | None
""" """
if not settings.enabled: if not settings.enabled:
@@ -140,10 +111,6 @@ def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
except ImportError: except ImportError:
logger.warning("Extra 'ai-litellm' requis pour le provider litellm") logger.warning("Extra 'ai-litellm' requis pour le provider litellm")
return None return None
if base_url is not None and (
_validate_base_url("litellm", base_url, settings.allow_insecure_http) is None
):
return None
return LiteLLMSynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model) return LiteLLMSynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model)
if settings.provider == "openai-compatible": if settings.provider == "openai-compatible":
@@ -154,8 +121,4 @@ def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
return None return None
return OpenAISynthesisProvider(api_key=settings.api_key, base_url=url, model=model) return OpenAISynthesisProvider(api_key=settings.api_key, base_url=url, model=model)
if base_url is not None and (
_validate_base_url("openai", base_url, settings.allow_insecure_http) is None
):
return None
return OpenAISynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model) return OpenAISynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model)
+23 -3
View File
@@ -52,10 +52,10 @@ def test_main_runs_composition_root_in_dry_run_with_requested_log_level(
runner.run.assert_called_once_with() runner.run.assert_called_once_with()
def test_main_preserves_configured_dry_run_and_returns_success_with_warnings( def test_main_preserves_configured_dry_run_and_returns_degraded_with_warnings(
mocker: MockerFixture, mocker: MockerFixture,
) -> None: ) -> None:
"""Sans option, la CLI préserve le dry-run configuré et accepte les avertissements.""" """Sans option, la CLI préserve le dry-run configuré et signale l'état dégradé."""
from pronote_sync.cli.main import main from pronote_sync.cli.main import main
settings = Settings(app=AppSettings(dry_run=True, log_level="WARNING")) settings = Settings(app=AppSettings(dry_run=True, log_level="WARNING"))
@@ -72,12 +72,32 @@ def test_main_preserves_configured_dry_run_and_returns_success_with_warnings(
exit_code = main([]) exit_code = main([])
assert exit_code == 0 assert exit_code == 2
assert setup_logging.call_args_list == [mocker.call("INFO"), mocker.call("WARNING")] assert setup_logging.call_args_list == [mocker.call("INFO"), mocker.call("WARNING")]
composition_root.assert_called_once_with(settings, dry_run=None) composition_root.assert_called_once_with(settings, dry_run=None)
runner.run.assert_called_once_with() runner.run.assert_called_once_with()
@pytest.mark.parametrize("step", ["caldav_sync", "send"])
def test_main_returns_degraded_code_for_caldav_or_xmpp_failure(
mocker: MockerFixture,
step: str,
) -> None:
"""Les échecs récupérables CalDAV et XMPP sont observables par le code 2."""
from pronote_sync.cli.main import main
settings = Settings()
mocker.patch("pronote_sync.cli.main.load_settings", return_value=settings)
runner = mocker.Mock()
runner.run.return_value = (
mocker.Mock(spec=PronoteData),
[PipelineWarning(f"Échec récupérable de {step}", step=step)],
)
mocker.patch("pronote_sync.cli.main.PipelineRunner.from_settings", return_value=runner)
assert main([]) == 2
def test_main_returns_failure_and_redacts_pipeline_secrets_at_debug_level( def test_main_returns_failure_and_redacts_pipeline_secrets_at_debug_level(
mocker: MockerFixture, mocker: MockerFixture,
capsys: pytest.CaptureFixture[str], capsys: pytest.CaptureFixture[str],
+3 -120
View File
@@ -10,10 +10,10 @@ from __future__ import annotations
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
import pytest import pytest
from pydantic import SecretStr, ValidationError from pydantic import SecretStr
from pronote_sync.config.env import load_settings from pronote_sync.config.env import load_settings
from pronote_sync.config.settings import AppSettings, PronoteSettings, Settings from pronote_sync.config.settings import PronoteSettings, Settings
if TYPE_CHECKING: if TYPE_CHECKING:
from _pytest.monkeypatch import MonkeyPatch from _pytest.monkeypatch import MonkeyPatch
@@ -207,121 +207,4 @@ def test_qr_pin_in_redaction_secrets(monkeypatch: MonkeyPatch) -> None:
assert "**********" in repr(settings.pronote.qr_pin) assert "**********" in repr(settings.pronote.qr_pin)
def test_sync_past_days_negative_direct_instantiation() -> None: # Ensure trailing newline
"""Vérifie que ``sync_past_days`` négatif lève ``ValidationError`` à l'instanciation.
:return: None
"""
with pytest.raises(ValidationError):
AppSettings(sync_past_days=-1)
def test_sync_future_days_negative_direct_instantiation() -> None:
"""Vérifie que ``sync_future_days`` négatif lève ``ValidationError`` à l'instanciation.
:return: None
"""
with pytest.raises(ValidationError):
AppSettings(sync_future_days=-1)
def test_sync_past_days_zero_accepted() -> None:
"""Vérifie que ``sync_past_days=0`` est accepté.
:return: None
"""
settings = AppSettings(sync_past_days=0)
assert settings.sync_past_days == 0
def test_sync_future_days_zero_accepted() -> None:
"""Vérifie que ``sync_future_days=0`` est accepté.
:return: None
"""
settings = AppSettings(sync_future_days=0)
assert settings.sync_future_days == 0
def test_sync_past_days_positive_accepted() -> None:
"""Vérifie que ``sync_past_days`` positif est accepté.
:return: None
"""
settings = AppSettings(sync_past_days=7)
assert settings.sync_past_days == 7
def test_sync_future_days_positive_accepted() -> None:
"""Vérifie que ``sync_future_days`` positif est accepté.
:return: None
"""
settings = AppSettings(sync_future_days=30)
assert settings.sync_future_days == 30
def test_sync_past_days_negative_env_loading(monkeypatch: MonkeyPatch) -> None:
"""Vérifie que ``SYNC_PAST_DAYS=-1`` lève ``ValidationError`` via chargement env.
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
:return: None
"""
monkeypatch.setenv("SYNC_PAST_DAYS", "-1")
with pytest.raises(ValidationError):
load_settings()
def test_sync_future_days_negative_env_loading(monkeypatch: MonkeyPatch) -> None:
"""Vérifie que ``SYNC_FUTURE_DAYS=-1`` lève ``ValidationError`` via chargement env.
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
:return: None
"""
monkeypatch.setenv("SYNC_FUTURE_DAYS", "-1")
with pytest.raises(ValidationError):
load_settings()
def test_sync_past_days_zero_env_loading(monkeypatch: MonkeyPatch) -> None:
"""Vérifie que ``SYNC_PAST_DAYS=0`` est accepté via chargement env.
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
:return: None
"""
monkeypatch.setenv("SYNC_PAST_DAYS", "0")
settings = load_settings()
assert settings.app.sync_past_days == 0
def test_sync_future_days_zero_env_loading(monkeypatch: MonkeyPatch) -> None:
"""Vérifie que ``SYNC_FUTURE_DAYS=0`` est accepté via chargement env.
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
:return: None
"""
monkeypatch.setenv("SYNC_FUTURE_DAYS", "0")
settings = load_settings()
assert settings.app.sync_future_days == 0
def test_sync_past_days_positive_env_loading(monkeypatch: MonkeyPatch) -> None:
"""Vérifie que ``SYNC_PAST_DAYS`` positif est accepté via chargement env.
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
:return: None
"""
monkeypatch.setenv("SYNC_PAST_DAYS", "7")
settings = load_settings()
assert settings.app.sync_past_days == 7
def test_sync_future_days_positive_env_loading(monkeypatch: MonkeyPatch) -> None:
"""Vérifie que ``SYNC_FUTURE_DAYS`` positif est accepté via chargement env.
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
:return: None
"""
monkeypatch.setenv("SYNC_FUTURE_DAYS", "30")
settings = load_settings()
assert settings.app.sync_future_days == 30
+1 -246
View File
@@ -621,17 +621,12 @@ def test_get_messages_degraded_on_error(
def test_get_informations_degraded_on_error( def test_get_informations_degraded_on_error(
mocker: pytest_mock.MockerFixture, mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
pronote_settings: PronoteSettings,
caplog: pytest.LogCaptureFixture,
) -> None: ) -> None:
"""Vérifie que get_informations retourne une liste vide en cas d'erreur réseau. """Vérifie que get_informations retourne une liste vide en cas d'erreur réseau.
Assert que le chemin d'erreur retourne toujours [] avec un log ERROR.
:param mocker: Fixture pytest-mock pour le mocking. :param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides. :param pronote_settings: Paramètres Pronote valides.
:param caplog: Fixture pour capturer les logs.
:return: None :return: None
""" """
mock_client = mocker.MagicMock() mock_client = mocker.MagicMock()
@@ -639,254 +634,14 @@ def test_get_informations_degraded_on_error(
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client) mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings) client = PronoteClient(pronote_settings)
with caplog.at_level(logging.ERROR, logger="pronote_sync.sources.pronote.client"):
messages = client.get_informations() messages = client.get_informations()
assert messages == [] assert messages == []
# Assert ERROR log is present
error_records = [r for r in caplog.records if r.levelno == logging.ERROR]
assert len(error_records) >= 1
assert any(
"Échec de la récupération des informations Pronote" in r.message for r in error_records
)
# --- QR code / token authentication tests --- # --- QR code / token authentication tests ---
# --- get_informations qr_token mode guard tests ---
def test_get_informations_skips_in_qr_token_mode(
mocker: pytest_mock.MockerFixture,
) -> None:
"""Vérifie que get_informations retourne [] immédiatement en mode qr_token.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
settings = PronoteSettings(
url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent="bordeaux",
account_type="parent",
auth_mode="qr_token",
)
client = PronoteClient(settings)
assert client._client is None
messages = client.get_informations()
assert messages == []
assert client._client is None
def test_get_informations_no_connect_in_qr_token_mode(
mocker: pytest_mock.MockerFixture,
) -> None:
"""Vérifie que _connect n'est jamais appelé en mode qr_token pour get_informations.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
settings = PronoteSettings(
url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent="bordeaux",
account_type="parent",
auth_mode="qr_token",
)
connect_spy = mocker.spy(PronoteClient, "_connect")
client = PronoteClient(settings)
messages = client.get_informations()
assert messages == []
connect_spy.assert_not_called()
def test_get_informations_no_information_and_surveys_in_qr_token_mode(
mocker: pytest_mock.MockerFixture,
) -> None:
"""Vérifie que information_and_surveys n'est jamais appelé en mode qr_token.
:param mocker: Fixture pytest-mock pour le mocking.
:return: None
"""
settings = PronoteSettings(
url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent="bordeaux",
account_type="parent",
auth_mode="qr_token",
)
mock_client = mocker.MagicMock()
mock_client.information_and_surveys = mocker.MagicMock()
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(settings)
messages = client.get_informations()
assert messages == []
mock_client.information_and_surveys.assert_not_called()
def test_get_informations_qr_token_no_side_effects(
mocker: pytest_mock.MockerFixture,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Prouve le contrat complet sans effet de bordure du chemin de saut qr_token.
Avec PronoteAuthState et _qr_token_operation_lock et _persist_credentials
mockés, assert que sur le saut : le gestionnaire de contexte de verrou n'est
PAS entré, auth-state load() n'est PAS appelé, _persist_credentials() n'est
PAS appelé, et l'export des credentials n'est PAS invoqué. Assert aussi que
self._client est inchangé.
:param mocker: Fixture pytest-mock pour le mocking.
:param caplog: Fixture pour capturer les logs.
:return: None
"""
auth_state = mocker.MagicMock(spec=PronoteAuthState)
auth_state.load = mocker.MagicMock()
auth_state.lock = mocker.MagicMock()
settings = PronoteSettings(
url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent="bordeaux",
account_type="parent",
auth_mode="qr_token",
)
client = PronoteClient(settings, auth_state=auth_state)
sentinel = MagicMock()
client._client = sentinel
persist_spy = mocker.spy(client, "_persist_credentials")
with caplog.at_level(logging.INFO, logger="pronote_sync.sources.pronote.client"):
messages = client.get_informations()
# Assert no side effects
assert messages == []
assert client._client is sentinel
assert isinstance(sentinel, MagicMock)
sentinel.export_credentials.assert_not_called()
persist_spy.assert_not_called()
auth_state.load.assert_not_called()
auth_state.lock.assert_not_called()
# Assert exactly one INFO log record with the exact message
info_records = [r for r in caplog.records if r.levelno == logging.INFO]
assert len(info_records) == 1
assert (
info_records[0].message
== "Récupération des informations Pronote ignorée : endpoint PageActualites "
"indisponible en mode d'authentification qr_token."
)
# Assert no secret sentinel appears in any log record
for record in caplog.records:
assert "testpass" not in record.message
assert "testuser" not in record.message
assert "pronote.example.com" not in record.message
def test_get_informations_logs_info_in_qr_token_mode(
mocker: pytest_mock.MockerFixture,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que get_informations log un message INFO exact en mode qr_token.
Assert exactement un enregistrement logging.INFO avec le message exact
(inspection de caplog.records, pas seulement caplog.text), et qu'aucune
sentinelle de secret n'apparaît.
:param mocker: Fixture pytest-mock pour le mocking.
:param caplog: Fixture pour capturer les logs.
:return: None
"""
settings = PronoteSettings(
url="https://pronote.example.com",
username="testuser",
password=SecretStr("testpass"),
ent="bordeaux",
account_type="parent",
auth_mode="qr_token",
)
with caplog.at_level(logging.INFO, logger="pronote_sync.sources.pronote.client"):
client = PronoteClient(settings)
messages = client.get_informations()
assert messages == []
# Assert exactly one INFO record with the exact message
info_records = [r for r in caplog.records if r.levelno == logging.INFO]
assert len(info_records) == 1
assert (
info_records[0].message
== "Récupération des informations Pronote ignorée : endpoint PageActualites "
"indisponible en mode d'authentification qr_token."
)
# Assert no secret sentinel appears in any record
for record in caplog.records:
assert "testpass" not in record.message
assert "testuser" not in record.message
assert "pronote.example.com" not in record.message
def test_get_informations_unchanged_in_password_mode(
mocker: pytest_mock.MockerFixture,
pronote_settings: PronoteSettings,
) -> None:
"""Vérifie que get_informations en mode password reste inchangé (régression).
Assert que _connect() A ÉTÉ appelé et information_and_surveys() A ÉTÉ appelé
(en cas de succès), en conservant les assertions de mappage existantes.
:param mocker: Fixture pytest-mock pour le mocking.
:param pronote_settings: Paramètres Pronote valides en mode password.
:return: None
"""
mock_client = mocker.MagicMock()
mock_info = mocker.MagicMock()
mock_info.id = "info-456"
mock_info.title = "Important Info"
mock_info.content.return_value = "Important content"
mock_info.author = "Admin"
mock_info.creation_date = datetime(2024, 9, 2, 14, 30, 0)
mock_info.read = False
mock_info.survey = True
mock_client.information_and_surveys.return_value = [mock_info]
# Patch _connect to return mock_client and track calls
connect_patch = mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
client = PronoteClient(pronote_settings)
messages = client.get_informations()
# Assert _connect() WAS called and information_and_surveys() WAS called
connect_patch.assert_called_once()
mock_client.information_and_surveys.assert_called_once()
# Retain existing mapping assertions
assert isinstance(messages, list)
assert len(messages) == 1
message = messages[0]
assert isinstance(message, Message)
assert message.id == "info-456"
assert message.type == MessageType.SURVEY
assert message.title == "Important Info"
assert message.content == "Important content"
assert message.author == "Admin"
assert message.date == datetime(2024, 9, 2, 14, 30, 0)
assert message.read is False
def test_connect_password_mode_unchanged( def test_connect_password_mode_unchanged(
mocker: pytest_mock.MockerFixture, mocker: pytest_mock.MockerFixture,
pronote_settings: PronoteSettings, pronote_settings: PronoteSettings,
+1 -466
View File
@@ -11,7 +11,7 @@ factory de sélection, en vérifiant :
from __future__ import annotations from __future__ import annotations
from datetime import date, datetime, time from datetime import date, datetime, time
from typing import TYPE_CHECKING, Any, Literal from typing import TYPE_CHECKING, Any
from unittest.mock import MagicMock from unittest.mock import MagicMock
import pytest import pytest
@@ -29,7 +29,6 @@ from pronote_sync.models.diff import AgendaChange, AgendaChangeType, AgendaDiff
from pronote_sync.models.message import Message, MessageType from pronote_sync.models.message import Message, MessageType
from pronote_sync.models.synthesis import SynthesisInput from pronote_sync.models.synthesis import SynthesisInput
from pronote_sync.synthesis import get_synthesis_provider from pronote_sync.synthesis import get_synthesis_provider
from pronote_sync.synthesis.litellm import LiteLLMSynthesisProvider
from pronote_sync.synthesis.openai import OpenAISynthesisProvider from pronote_sync.synthesis.openai import OpenAISynthesisProvider
from pronote_sync.synthesis.provider import SynthesisProvider from pronote_sync.synthesis.provider import SynthesisProvider
@@ -873,70 +872,6 @@ def test_openai_compatible_valid_https_url_accepted() -> None:
assert isinstance(result, OpenAISynthesisProvider) assert isinstance(result, OpenAISynthesisProvider)
def test_openai_compatible_url_returned_unchanged() -> None:
"""Vérifie que l'URL est retournée strictement inchangée, sans manipulation de /v1."""
custom_url = "https://api.example.com/custom/path?query=value"
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai-compatible",
base_url=custom_url,
model="test-model",
)
result = get_synthesis_provider(settings)
assert isinstance(result, OpenAISynthesisProvider)
# OpenAI SDK appends a trailing slash to base_url, so we check the string representation
assert str(result._client.base_url).rstrip("/") == custom_url
def test_openai_url_returned_unchanged() -> None:
"""Vérifie que l'URL est retournée strictement inchangée pour openai."""
custom_url = "https://api.example.com/custom/path?query=value"
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url=custom_url,
)
result = get_synthesis_provider(settings)
assert isinstance(result, OpenAISynthesisProvider)
# OpenAI SDK appends a trailing slash to base_url, so we check the string representation
assert str(result._client.base_url).rstrip("/") == custom_url
def test_litellm_url_returned_unchanged() -> None:
"""Vérifie que l'URL est retournée strictement inchangée pour litellm."""
pytest.importorskip("litellm")
custom_url = "https://api.example.com/custom/path?query=value"
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url=custom_url,
)
result = get_synthesis_provider(settings)
assert result is not None
assert isinstance(result, LiteLLMSynthesisProvider)
assert result._base_url == custom_url
def test_openai_compatible_malformed_port_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'un port malformé est refusé pour openai-compatible."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai-compatible",
base_url="https://host:bad/v1",
model="test-model",
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL invalide" in caplog.text
assert "openai-compatible" in caplog.text
def test_openai_compatible_http_refused_by_default( def test_openai_compatible_http_refused_by_default(
caplog: pytest.LogCaptureFixture, caplog: pytest.LogCaptureFixture,
) -> None: ) -> None:
@@ -952,7 +887,6 @@ def test_openai_compatible_http_refused_by_default(
result = get_synthesis_provider(settings) result = get_synthesis_provider(settings)
assert result is None assert result is None
assert "URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true" in caplog.text assert "URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true" in caplog.text
assert "openai-compatible" in caplog.text
def test_openai_compatible_http_accepted_with_allow_insecure_http() -> None: def test_openai_compatible_http_accepted_with_allow_insecure_http() -> None:
@@ -983,7 +917,6 @@ def test_openai_compatible_credentials_in_url_refused(
result = get_synthesis_provider(settings) result = get_synthesis_provider(settings)
assert result is None assert result is None
assert "Credentials dans l'URL refusés" in caplog.text assert "Credentials dans l'URL refusés" in caplog.text
assert "openai-compatible" in caplog.text
def test_openai_compatible_sensitive_query_params_refused( def test_openai_compatible_sensitive_query_params_refused(
@@ -1000,24 +933,6 @@ def test_openai_compatible_sensitive_query_params_refused(
result = get_synthesis_provider(settings) result = get_synthesis_provider(settings)
assert result is None assert result is None
assert "Paramètres sensibles dans l'URL refusés" in caplog.text assert "Paramètres sensibles dans l'URL refusés" in caplog.text
assert "openai-compatible" in caplog.text
def test_openai_compatible_sensitive_query_params_valueless_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que les query params sensibles sans valeur sont refusés pour openai-compatible."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai-compatible",
base_url="https://host/v1?token",
model="test-model",
)
result = get_synthesis_provider(settings)
assert result is None
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
assert "openai-compatible" in caplog.text
def test_openai_compatible_connection_error_returns_none( def test_openai_compatible_connection_error_returns_none(
@@ -1076,383 +991,3 @@ def test_openai_compatible_factory_no_network_calls(
assert isinstance(result, OpenAISynthesisProvider) assert isinstance(result, OpenAISynthesisProvider)
mock_get.assert_not_called() mock_get.assert_not_called()
mock_post.assert_not_called() mock_post.assert_not_called()
# --- Tests de validation AI_BASE_URL pour openai et litellm (Issue #17) ---
def test_openai_valid_https_base_url_accepted() -> None:
"""Vérifie qu'une URL HTTPS valide est acceptée pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="https://api.openai.com/v1",
)
result = get_synthesis_provider(settings)
assert isinstance(result, OpenAISynthesisProvider)
def test_openai_http_refused_by_default(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que HTTP est refusé par défaut pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="http://127.0.0.1:11434/v1",
allow_insecure_http=False,
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true" in caplog.text
assert "openai" in caplog.text
def test_openai_http_accepted_with_allow_insecure_http() -> None:
"""Vérifie que HTTP est accepté avec allow_insecure_http=True pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="http://127.0.0.1:11434/v1",
allow_insecure_http=True,
)
result = get_synthesis_provider(settings)
assert isinstance(result, OpenAISynthesisProvider)
def test_openai_credentials_in_url_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que les credentials dans l'URL sont refusés pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="https://user:pass@host/v1", # pragma: allowlist secret
)
result = get_synthesis_provider(settings)
assert result is None
assert "Credentials dans l'URL refusés" in caplog.text
assert "openai" in caplog.text
def test_openai_sensitive_query_params_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que les query params sensibles sont refusés pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="https://host/v1?token=secret",
)
result = get_synthesis_provider(settings)
assert result is None
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
assert "openai" in caplog.text
def test_openai_sensitive_query_params_valueless_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que les query params sensibles sans valeur sont refusés pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="https://host/v1?token",
)
result = get_synthesis_provider(settings)
assert result is None
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
assert "openai" in caplog.text
def test_openai_malformed_url_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'une URL malformée est refusée pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="not-a-valid-url",
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL invalide" in caplog.text
assert "openai" in caplog.text
def test_openai_no_hostname_url_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'une URL sans hostname est refusée pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="https:///v1",
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL sans hostname" in caplog.text
assert "openai" in caplog.text
def test_openai_malformed_port_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'un port malformé est refusé pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="https://host:bad/v1",
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL invalide" in caplog.text
assert "openai" in caplog.text
def test_openai_no_network_calls_during_validation(
mocker: MockerFixture,
) -> None:
"""Vérifie qu'aucun appel réseau n'est effectué pendant la validation pour openai."""
mock_get = mocker.patch("requests.get")
mock_post = mocker.patch("requests.post")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="https://api.example.com/v1",
)
result = get_synthesis_provider(settings)
assert isinstance(result, OpenAISynthesisProvider)
mock_get.assert_not_called()
mock_post.assert_not_called()
def test_openai_sentinel_key_not_in_logs(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'une clé sentinelle est absente des logs pour openai."""
sentinel = "sk-SENTINEL-OPENAI-BASE-URL-12345"
settings = AISettings(
enabled=True,
api_key=SecretStr(sentinel),
provider="openai",
base_url="https://user:pass@host/v1", # pragma: allowlist secret
)
result = get_synthesis_provider(settings)
assert result is None
assert sentinel not in caplog.text
def test_litellm_valid_https_base_url_accepted() -> None:
"""Vérifie qu'une URL HTTPS valide est acceptée pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="https://api.litellm.ai/v1",
)
result = get_synthesis_provider(settings)
assert result is not None
def test_litellm_http_refused_by_default(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que HTTP est refusé par défaut pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="http://127.0.0.1:11434/v1",
allow_insecure_http=False,
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true" in caplog.text
assert "litellm" in caplog.text
def test_litellm_http_accepted_with_allow_insecure_http() -> None:
"""Vérifie que HTTP est accepté avec allow_insecure_http=True pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="http://127.0.0.1:11434/v1",
allow_insecure_http=True,
)
result = get_synthesis_provider(settings)
assert result is not None
def test_litellm_credentials_in_url_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que les credentials dans l'URL sont refusés pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="https://user:pass@host/v1", # pragma: allowlist secret
)
result = get_synthesis_provider(settings)
assert result is None
assert "Credentials dans l'URL refusés" in caplog.text
assert "litellm" in caplog.text
def test_litellm_sensitive_query_params_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que les query params sensibles sont refusés pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="https://host/v1?token=secret",
)
result = get_synthesis_provider(settings)
assert result is None
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
assert "litellm" in caplog.text
def test_litellm_sensitive_query_params_valueless_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que les query params sensibles sans valeur sont refusés pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="https://host/v1?token",
)
result = get_synthesis_provider(settings)
assert result is None
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
assert "litellm" in caplog.text
def test_litellm_malformed_url_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'une URL malformée est refusée pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="not-a-valid-url",
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL invalide" in caplog.text
assert "litellm" in caplog.text
def test_litellm_no_hostname_url_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'une URL sans hostname est refusée pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="https:///v1",
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL sans hostname" in caplog.text
assert "litellm" in caplog.text
def test_litellm_malformed_port_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'un port malformé est refusé pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="https://host:bad/v1",
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL invalide" in caplog.text
assert "litellm" in caplog.text
def test_litellm_no_network_calls_during_validation(
mocker: MockerFixture,
) -> None:
"""Vérifie qu'aucun appel réseau n'est effectué pendant la validation pour litellm."""
pytest.importorskip("litellm")
mock_get = mocker.patch("requests.get")
mock_post = mocker.patch("requests.post")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="https://api.example.com/v1",
)
result = get_synthesis_provider(settings)
assert result is not None
mock_get.assert_not_called()
mock_post.assert_not_called()
def test_litellm_sentinel_key_not_in_logs(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'une clé sentinelle est absente des logs pour litellm."""
pytest.importorskip("litellm")
sentinel = "sk-SENTINEL-LITELLM-BASE-URL-67890"
settings = AISettings(
enabled=True,
api_key=SecretStr(sentinel),
provider="litellm",
base_url="https://user:pass@host/v1", # pragma: allowlist secret
)
result = get_synthesis_provider(settings)
assert result is None
assert sentinel not in caplog.text
@pytest.mark.parametrize("provider", ["openai", "litellm", "openai-compatible"])
def test_all_providers_http_refused_same_warning(
provider: Literal["openai", "litellm", "openai-compatible"], caplog: pytest.LogCaptureFixture
) -> None:
"""Vérifie que tous les providers émettent le même message d'avertissement pour HTTP refusé."""
if provider == "litellm":
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider=provider,
base_url="http://127.0.0.1:11434/v1",
allow_insecure_http=False,
model="test-model" if provider == "openai-compatible" else None,
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true" in caplog.text