Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4ad62ad50f | ||
|
|
2d55ac8d2b |
+3
-8
@@ -23,6 +23,8 @@ PRONOTE_AUTH_MODE=password
|
||||
# PRONOTE_AUTH_MODE=qr_token
|
||||
# PRONOTE_QR_CODE_FILE=/path/to/qr_code.json
|
||||
# PRONOTE_QR_PIN=1234
|
||||
# PIN de second facteur du compte Pronote, distinct du PIN de déchiffrement du QR code
|
||||
# PRONOTE_ACCOUNT_PIN=
|
||||
|
||||
# --- CalDAV ---
|
||||
CALDAV_URL=https://caldav.example.com/calendars/user/pronote/
|
||||
@@ -32,7 +34,7 @@ CALDAV_CALENDAR_PATH=/pronote-sync/
|
||||
# Autoriser HTTP (non-HTTPS) pour un serveur CalDAV local (localhost uniquement)
|
||||
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_FUTURE_DAYS=30
|
||||
|
||||
@@ -66,13 +68,6 @@ AI_BASE_URL=https://api.openai.com/v1
|
||||
# AI_API_KEY=
|
||||
# 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)
|
||||
# AI_PROVIDER=openai-compatible
|
||||
# AI_BASE_URL=https://openrouter.ai/api/v1
|
||||
|
||||
+2
-2
@@ -140,7 +140,7 @@
|
||||
"filename": "GUIDE_DEV_PYTHON.md",
|
||||
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
|
||||
"is_verified": false,
|
||||
"line_number": 5186
|
||||
"line_number": 5125
|
||||
}
|
||||
],
|
||||
"tests/unit/test_caldav_gateway.py": [
|
||||
@@ -185,5 +185,5 @@
|
||||
}
|
||||
]
|
||||
},
|
||||
"generated_at": "2026-09-12T17:57:39Z"
|
||||
"generated_at": "2026-09-12T12:04:07Z"
|
||||
}
|
||||
|
||||
@@ -159,12 +159,9 @@ pronote-sync --dry-run
|
||||
- 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
|
||||
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
|
||||
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
|
||||
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.
|
||||
(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,
|
||||
informations) pour garantir la persistance du token valide.
|
||||
- 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.
|
||||
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`)
|
||||
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.
|
||||
- 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`).
|
||||
Son contenu (token vivant) ne doit jamais apparaître dans les logs, les messages d'erreur ou
|
||||
les notifications XMPP.
|
||||
|
||||
+5
-1
@@ -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_ENT` rendu optionnel pour les connexions pronotepy directes
|
||||
- `.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
|
||||
|
||||
- Wiki `GuidePronote` enrichi : section "Quand l'ENT est obligatoire" (EduConnect/HubEduConnect), exemple Bordeaux
|
||||
- `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
|
||||
|
||||
- 694 tests passés, couverture 94.93%
|
||||
|
||||
+46
-107
@@ -299,18 +299,18 @@ d'un besoin réel et testé.
|
||||
| `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_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_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_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. | `30` | `int` |
|
||||
| `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`|
|
||||
| `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`|
|
||||
| `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_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_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` |
|
||||
| `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/"`.
|
||||
> `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.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
|
||||
from typing import Literal
|
||||
@@ -421,8 +421,8 @@ class CalDAVSettings(BaseSettings):
|
||||
password: SecretStr | None = None
|
||||
calendar_path: str = "/pronote-sync/"
|
||||
allow_insecure_http: bool = False
|
||||
sync_past_days: int = Field(default=7, ge=0)
|
||||
sync_future_days: int = Field(default=30, ge=0)
|
||||
sync_past_days: int = 7
|
||||
sync_future_days: int = 30
|
||||
|
||||
|
||||
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
|
||||
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é à
|
||||
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`.
|
||||
|
||||
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
|
||||
`qr_token`), puis persistance des credentials actualisées. En mode `qr_token`, `get_informations()`
|
||||
est ignorée (retour immédiat `[]` sans connexion ni verrou) : elle n'acquiert pas le verrou et ne
|
||||
déclenche aucune persistance. Une tentative concurrente échoue immédiatement avec une erreur d'état
|
||||
d'authentification expurgée ; elle ne patiente pas et ne relance pas l'authentification. Le contenu
|
||||
du token, le PIN et les autres credentials ne sont jamais inclus dans les logs ni dans ce message
|
||||
d'erreur.
|
||||
enrôlement QR initial, opération de données (agenda, devoirs, messages ou informations), puis
|
||||
persistance des credentials actualisées. Une tentative concurrente échoue immédiatement avec une
|
||||
erreur d'état d'authentification expurgée ; elle ne patiente pas et ne relance pas
|
||||
l'authentification. Le contenu 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
|
||||
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`` |
|
||||
| ``litellm`` | Bibliothèque LiteLLM embarquée | ``LiteLLMSynthesisProvider`` |
|
||||
|
||||
Pour le provider ``openai-compatible``, ``AI_BASE_URL`` et ``AI_MODEL`` sont requis ; pour
|
||||
``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 :
|
||||
Pour le provider ``openai-compatible``, la validation de la configuration est stricte :
|
||||
|
||||
- URL parsable par ``urlparse`` (``ValueError`` ou schéma vide → refusée) et hostname non vide.
|
||||
- Schéma limité à ``http``/``https`` ; ``http`` refusé sauf si ``AI_ALLOW_INSECURE_HTTP=true``.
|
||||
- Credentials dans l'URL (``user:pass@host``) refusés.
|
||||
- Paramètres sensibles dans la *query string* refusés, y compris sans valeur
|
||||
(``token``, ``key``, ``api_key``, ``secret``, ``password``, ``auth``).
|
||||
- ``AI_BASE_URL`` est requis.
|
||||
- ``AI_MODEL`` est requis et ne doit pas être vide.
|
||||
- ``AI_API_KEY`` est requis (MVP).
|
||||
- L'URL doit utiliser le schéma ``https`` sauf si ``AI_ALLOW_INSECURE_HTTP=true``.
|
||||
- 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.
|
||||
- En cas de violation, un avertissement est journalisé (URL masquée via ``redact_url``) et la
|
||||
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`) |
|
||||
- Si la configuration est incomplète, la factory retourne ``None`` avec un avertissement (mode dégradé).
|
||||
|
||||
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__)
|
||||
|
||||
|
||||
_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(
|
||||
url: str | None, model: str | None, allow_insecure_http: bool
|
||||
) -> str | None:
|
||||
"""Valide la configuration du provider ``openai-compatible``."""
|
||||
if not url:
|
||||
logger.warning("URL de base requise pour le provider openai-compatible")
|
||||
if not url or not model:
|
||||
return None
|
||||
if not model:
|
||||
logger.warning("Modèle requis pour le provider openai-compatible")
|
||||
try:
|
||||
parsed = urlparse(url)
|
||||
except ValueError:
|
||||
logger.warning("URL invalide : %s", redact_url(url))
|
||||
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:
|
||||
@@ -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é
|
||||
API n'est configurée. Pour le provider ``litellm``, le paquet ``litellm``
|
||||
(extra ``ai-litellm``) est requis : s'il est absent, un avertissement est
|
||||
journalisé et ``None`` est retourné. Pour ``openai`` et ``litellm``,
|
||||
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`.
|
||||
journalisé et ``None`` est retourné.
|
||||
|
||||
:param settings: Paramètres IA.
|
||||
: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:
|
||||
logger.warning("Extra 'ai-litellm' requis pour le provider litellm")
|
||||
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)
|
||||
|
||||
if settings.provider == "openai-compatible":
|
||||
@@ -4068,10 +4011,6 @@ def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
|
||||
return None
|
||||
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)
|
||||
```
|
||||
|
||||
|
||||
+1
-1
@@ -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`)
|
||||
|
||||
- **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_ENABLED`, `BLOG_RSS_URL`
|
||||
|
||||
@@ -66,6 +66,11 @@ processus. Le mode `PRONOTE_AUTH_MODE=qr_token` est incompatible avec cette gara
|
||||
le refuse avant toute connexion afin de ne pas désynchroniser le token local du token distant.
|
||||
Le dry-run ne remplace pas une vérification des paramètres réellement chargés.
|
||||
|
||||
En mode `PRONOTE_AUTH_MODE=qr_token`, `PRONOTE_QR_PIN` déchiffre le QR code
|
||||
exporté depuis le site web Pronote. Si le compte exige un second facteur,
|
||||
configurez aussi `PRONOTE_ACCOUNT_PIN` avec le PIN du compte. Ce PIN est
|
||||
transmis uniquement à `pronotepy` lors de l'enrôlement QR et des connexions par
|
||||
token ; il n'est jamais écrit dans `.pronote_auth_state.json` ni dans les logs.
|
||||
En mode `PRONOTE_AUTH_MODE=qr_token`, le fichier
|
||||
`.pronote_auth_state.json` et son verrou frère sont créés dans le répertoire
|
||||
de travail du service (par exemple `/var/lib/pronote-sync`) avec le mode
|
||||
|
||||
@@ -47,6 +47,7 @@ class PronoteSettings(BaseSettings):
|
||||
auth_mode: Literal["password", "qr_token"] = "password"
|
||||
qr_code_file: str | None = None
|
||||
qr_pin: SecretStr | None = None
|
||||
account_pin: SecretStr | None = None
|
||||
|
||||
@field_serializer("ical_url")
|
||||
def _serialize_ical_url(self, value: SecretStr | None) -> str | None:
|
||||
@@ -72,6 +73,18 @@ class PronoteSettings(BaseSettings):
|
||||
return None
|
||||
return "**********"
|
||||
|
||||
@field_serializer("account_pin")
|
||||
def _serialize_account_pin(self, value: SecretStr | None) -> str | None:
|
||||
"""Masque le PIN du compte lors de la sérialisation.
|
||||
|
||||
:param value: Valeur du PIN de second facteur du compte.
|
||||
:return: ``"**********"`` si la valeur est définie, ``None`` sinon.
|
||||
:rtype: str | None
|
||||
"""
|
||||
if value is None:
|
||||
return None
|
||||
return "**********"
|
||||
|
||||
|
||||
class CalDAVSettings(BaseSettings):
|
||||
"""Paramètres d'accès au serveur CalDAV de destination.
|
||||
@@ -293,8 +306,8 @@ class AppSettings(BaseSettings):
|
||||
school_holidays_path: str | None = None
|
||||
theoretical_week_anchor_date: date | None = None
|
||||
theoretical_week_anchor_type: Literal["even", "odd"] | None = None
|
||||
sync_past_days: int = Field(default=7, ge=0)
|
||||
sync_future_days: int = Field(default=30, ge=0)
|
||||
sync_past_days: int = 7
|
||||
sync_future_days: int = 30
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
@@ -328,6 +341,7 @@ class Settings(BaseSettings):
|
||||
self.pronote.ical_url,
|
||||
self.pronote.password,
|
||||
self.pronote.qr_pin,
|
||||
self.pronote.account_pin,
|
||||
self.caldav.url,
|
||||
self.caldav.password,
|
||||
self.xmpp.password,
|
||||
|
||||
@@ -119,6 +119,9 @@ def _collect_auth_secrets(client: PronoteClient) -> list[str]:
|
||||
# PIN QR
|
||||
if settings.qr_pin is not None:
|
||||
secrets.append(settings.qr_pin.get_secret_value())
|
||||
# PIN de second facteur du compte
|
||||
if settings.account_pin is not None:
|
||||
secrets.append(settings.account_pin.get_secret_value())
|
||||
# Contenu du fichier QR (jeton, login, url)
|
||||
if settings.qr_code_file is not None:
|
||||
try:
|
||||
@@ -337,7 +340,12 @@ class PronoteClient:
|
||||
creds = self._auth_state.load()
|
||||
if creds is not None:
|
||||
try:
|
||||
client = client_class.token_login(**creds)
|
||||
account_pin = (
|
||||
self._settings.account_pin.get_secret_value()
|
||||
if self._settings.account_pin is not None
|
||||
else None
|
||||
)
|
||||
client = client_class.token_login(**creds, account_pin=account_pin)
|
||||
if client.logged_in:
|
||||
self._client = client
|
||||
self._persist_credentials()
|
||||
@@ -421,10 +429,16 @@ class PronoteClient:
|
||||
app_uuid = f"pronote-sync-{uuid4().hex}"
|
||||
|
||||
try:
|
||||
account_pin = (
|
||||
self._settings.account_pin.get_secret_value()
|
||||
if self._settings.account_pin is not None
|
||||
else None
|
||||
)
|
||||
client = client_class.qrcode_login(
|
||||
qr_code=qr_data,
|
||||
pin=pin_value,
|
||||
uuid=app_uuid,
|
||||
account_pin=account_pin,
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
@@ -488,23 +502,9 @@ class PronoteClient:
|
||||
Chaque entrée est mappée sur un modèle :class:`Message` de type
|
||||
``SURVEY`` si c'est un sondage, ``INFORMATION`` sinon.
|
||||
|
||||
En mode ``qr_token``, la récupération est ignorée sans connexion ni
|
||||
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``.
|
||||
:return: Liste des informations et sondages ; vide en cas d'erreur.
|
||||
: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():
|
||||
try:
|
||||
client = self._connect()
|
||||
|
||||
@@ -15,83 +15,20 @@ logger = logging.getLogger(__name__)
|
||||
__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(
|
||||
url: str | None, model: str | None, allow_insecure_http: bool
|
||||
) -> str | None:
|
||||
"""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
|
||||
à ``openai-compatible``), puis délègue les règles structurelles
|
||||
partagées à :func:`_validate_base_url`. En cas d'échec, un
|
||||
avertissement est journalisé et ``None`` est retourné : la synthèse IA
|
||||
se dégrade silencieusement, sans jamais lever d'exception.
|
||||
Vérifie la présence de l'URL de base et du modèle, le schéma de l'URL
|
||||
(HTTPS obligatoire, HTTP accepté uniquement si ``allow_insecure_http``
|
||||
vaut ``True``), la présence d'un hostname non vide, l'absence
|
||||
d'identifiants dans le netloc et de paramètres sensibles dans la
|
||||
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 model: Identifiant du modèle à utiliser.
|
||||
@@ -106,7 +43,42 @@ def _validate_openai_compatible_config(
|
||||
if not model:
|
||||
logger.warning("Modèle requis pour le provider openai-compatible")
|
||||
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:
|
||||
@@ -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é
|
||||
API n'est configurée. Pour le provider ``litellm``, le paquet ``litellm``
|
||||
(extra ``ai-litellm``) est requis : s'il est absent, un avertissement est
|
||||
journalisé et ``None`` est retourné. Pour ``openai`` et ``litellm``,
|
||||
une ``base_url`` éventuelle est validée par :func:`_validate_base_url` ;
|
||||
pour le provider ``openai-compatible``, la configuration (URL de base
|
||||
et modèle) est validée par :func:`_validate_openai_compatible_config` ;
|
||||
en cas de rejet, ``None`` est retourné avec un avertissement.
|
||||
journalisé et ``None`` est retourné. Pour le provider
|
||||
``openai-compatible``, la configuration (URL de base et modèle) est
|
||||
validée par :func:`_validate_openai_compatible_config` ; en cas de
|
||||
rejet, ``None`` est retourné avec un avertissement.
|
||||
|
||||
:param settings: Paramètres IA.
|
||||
: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
|
||||
"""
|
||||
if not settings.enabled:
|
||||
@@ -140,10 +111,6 @@ def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
|
||||
except ImportError:
|
||||
logger.warning("Extra 'ai-litellm' requis pour le provider litellm")
|
||||
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)
|
||||
|
||||
if settings.provider == "openai-compatible":
|
||||
@@ -154,8 +121,4 @@ def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
|
||||
return None
|
||||
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)
|
||||
|
||||
+13
-113
@@ -10,10 +10,10 @@ from __future__ import annotations
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
from pydantic import SecretStr, ValidationError
|
||||
from pydantic import SecretStr
|
||||
|
||||
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:
|
||||
from _pytest.monkeypatch import MonkeyPatch
|
||||
@@ -207,121 +207,21 @@ def test_qr_pin_in_redaction_secrets(monkeypatch: MonkeyPatch) -> None:
|
||||
assert "**********" in repr(settings.pronote.qr_pin)
|
||||
|
||||
|
||||
def test_sync_past_days_negative_direct_instantiation() -> None:
|
||||
"""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.
|
||||
def test_account_pin_loaded_as_secretstr_and_redacted(monkeypatch: MonkeyPatch) -> None:
|
||||
"""Vérifie que ``PRONOTE_ACCOUNT_PIN`` est secret et expurgé.
|
||||
|
||||
: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")
|
||||
monkeypatch.setenv("PRONOTE_ACCOUNT_PIN", "account-pin-42")
|
||||
settings = load_settings()
|
||||
assert settings.app.sync_past_days == 0
|
||||
|
||||
assert isinstance(settings.pronote.account_pin, SecretStr)
|
||||
assert settings.pronote.account_pin.get_secret_value() == "account-pin-42"
|
||||
assert settings.pronote.account_pin in settings.redaction_secrets()
|
||||
assert "account-pin-42" not in str(settings)
|
||||
assert "account-pin-42" not in settings.model_dump_json()
|
||||
assert "**********" in settings.model_dump_json()
|
||||
|
||||
|
||||
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
|
||||
# Ensure trailing newline
|
||||
|
||||
@@ -621,17 +621,12 @@ def test_get_messages_degraded_on_error(
|
||||
|
||||
|
||||
def test_get_informations_degraded_on_error(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
pronote_settings: PronoteSettings,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
mocker: pytest_mock.MockerFixture, pronote_settings: PronoteSettings
|
||||
) -> None:
|
||||
"""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 pronote_settings: Paramètres Pronote valides.
|
||||
:param caplog: Fixture pour capturer les logs.
|
||||
:return: None
|
||||
"""
|
||||
mock_client = mocker.MagicMock()
|
||||
@@ -639,254 +634,14 @@ def test_get_informations_degraded_on_error(
|
||||
mocker.patch.object(PronoteClient, "_connect", return_value=mock_client)
|
||||
|
||||
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 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 ---
|
||||
|
||||
|
||||
# --- 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(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
pronote_settings: PronoteSettings,
|
||||
@@ -919,8 +674,10 @@ def test_connect_password_mode_unchanged(
|
||||
assert mock_client_class.call_count == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize("account_pin", [None, "account-pin-42"])
|
||||
def test_connect_qr_token_with_persisted_creds(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
account_pin: str | None,
|
||||
) -> None:
|
||||
"""Vérifie le login par token persisté en mode qr_token.
|
||||
|
||||
@@ -928,6 +685,7 @@ def test_connect_qr_token_with_persisted_creds(
|
||||
``token_login`` et le token rotate est resauvegardé.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param account_pin: PIN de second facteur facultatif.
|
||||
:return: None
|
||||
"""
|
||||
creds = {
|
||||
@@ -952,23 +710,29 @@ def test_connect_qr_token_with_persisted_creds(
|
||||
ent=None,
|
||||
account_type="parent",
|
||||
auth_mode="qr_token",
|
||||
account_pin=SecretStr(account_pin) if account_pin is not None else None,
|
||||
)
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
connected = client._connect()
|
||||
|
||||
assert connected is mock_client
|
||||
pronotepy.ParentClient.token_login.assert_called_once_with(**creds) # type: ignore[attr-defined]
|
||||
pronotepy.ParentClient.token_login.assert_called_once_with( # type: ignore[attr-defined]
|
||||
**creds, account_pin=account_pin
|
||||
)
|
||||
auth_state.save.assert_called_once_with(rotated_creds)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("account_pin", [None, "account-pin-42"])
|
||||
def test_connect_qr_token_no_creds_with_qr_code(
|
||||
mocker: pytest_mock.MockerFixture,
|
||||
tmp_path: Path,
|
||||
account_pin: str | None,
|
||||
) -> None:
|
||||
"""Vérifie l'enrôlement initial par QR code quand aucun token n'est persisté.
|
||||
|
||||
:param mocker: Fixture pytest-mock pour le mocking.
|
||||
:param tmp_path: Répertoire temporaire de test.
|
||||
:param account_pin: PIN de second facteur facultatif.
|
||||
:return: None
|
||||
"""
|
||||
qr_file = tmp_path / "qr_code.json"
|
||||
@@ -1005,6 +769,7 @@ def test_connect_qr_token_no_creds_with_qr_code(
|
||||
auth_mode="qr_token",
|
||||
qr_code_file=str(qr_file),
|
||||
qr_pin=SecretStr("123456"),
|
||||
account_pin=SecretStr(account_pin) if account_pin is not None else None,
|
||||
)
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
connected = client._connect()
|
||||
@@ -1014,6 +779,7 @@ def test_connect_qr_token_no_creds_with_qr_code(
|
||||
qrcode_login.assert_called_once() # type: ignore[attr-defined]
|
||||
kwargs = qrcode_login.call_args.kwargs # type: ignore[attr-defined]
|
||||
assert kwargs["pin"] == "123456"
|
||||
assert kwargs["account_pin"] == account_pin
|
||||
assert kwargs["qr_code"] == {
|
||||
"login": "testuser",
|
||||
"jeton": "qr-jeton",
|
||||
@@ -1412,6 +1178,7 @@ def test_no_raw_secrets_in_logs(
|
||||
sentinel_token = "SENTINEL_RAW_TOKEN_ALPHA"
|
||||
sentinel_pin = "SENTINEL_RAW_PIN_BRAVO"
|
||||
sentinel_jeton = "SENTINEL_RAW_JETON_CHARLIE"
|
||||
sentinel_account_pin = "SENTINEL_RAW_ACCOUNT_PIN_DELTA"
|
||||
|
||||
qr_file = tmp_path / "qr_code.json"
|
||||
qr_file.write_text(
|
||||
@@ -1435,7 +1202,8 @@ def test_no_raw_secrets_in_logs(
|
||||
mocker.patch(
|
||||
"pronotepy.ParentClient.token_login",
|
||||
side_effect=pronotepy.PronoteAPIError(
|
||||
f"login refusé {sentinel_token} puis {sentinel_pin} puis {sentinel_jeton}"
|
||||
f"login refusé {sentinel_token} puis {sentinel_pin} puis {sentinel_jeton} "
|
||||
f"puis {sentinel_account_pin}"
|
||||
),
|
||||
)
|
||||
|
||||
@@ -1448,6 +1216,7 @@ def test_no_raw_secrets_in_logs(
|
||||
auth_mode="qr_token",
|
||||
qr_code_file=str(qr_file),
|
||||
qr_pin=SecretStr(sentinel_pin),
|
||||
account_pin=SecretStr(sentinel_account_pin),
|
||||
)
|
||||
client = PronoteClient(settings, auth_state=auth_state)
|
||||
|
||||
@@ -1459,10 +1228,12 @@ def test_no_raw_secrets_in_logs(
|
||||
assert sentinel_token not in message
|
||||
assert sentinel_pin not in message
|
||||
assert sentinel_jeton not in message
|
||||
assert sentinel_account_pin not in message
|
||||
assert caplog.text
|
||||
assert sentinel_token not in caplog.text
|
||||
assert sentinel_pin not in caplog.text
|
||||
assert sentinel_jeton not in caplog.text
|
||||
assert sentinel_account_pin not in caplog.text
|
||||
|
||||
|
||||
# --- Persistence of credentials after data operations ---
|
||||
|
||||
@@ -11,7 +11,7 @@ factory de sélection, en vérifiant :
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date, datetime, time
|
||||
from typing import TYPE_CHECKING, Any, Literal
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
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.synthesis import SynthesisInput
|
||||
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.provider import SynthesisProvider
|
||||
|
||||
@@ -873,70 +872,6 @@ def test_openai_compatible_valid_https_url_accepted() -> None:
|
||||
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(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
@@ -952,7 +887,6 @@ def test_openai_compatible_http_refused_by_default(
|
||||
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-compatible" in caplog.text
|
||||
|
||||
|
||||
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)
|
||||
assert result is None
|
||||
assert "Credentials dans l'URL refusés" in caplog.text
|
||||
assert "openai-compatible" in caplog.text
|
||||
|
||||
|
||||
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)
|
||||
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_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(
|
||||
@@ -1076,383 +991,3 @@ def test_openai_compatible_factory_no_network_calls(
|
||||
assert isinstance(result, OpenAISynthesisProvider)
|
||||
mock_get.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
|
||||
|
||||
Reference in New Issue
Block a user