refactor(config): migrer RSS et Pronote vers l'endpoint commun et durcir le contrat #64
@@ -26,17 +26,23 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
|
|||||||
|
|
||||||
from pronote_sync.utils.redaction import redact_url
|
from pronote_sync.utils.redaction import redact_url
|
||||||
|
|
||||||
_EXTERNAL_ENDPOINT_SCHEMES: frozenset[str] = frozenset({"file", "http", "https"})
|
_EXTERNAL_ENDPOINT_SCHEMES: frozenset[str] = frozenset({"http", "https"})
|
||||||
_LOOPBACK_HOSTS: frozenset[str] = frozenset({"localhost", "127.0.0.1", "::1"})
|
_LOOPBACK_HOSTS: frozenset[str] = frozenset({"localhost", "127.0.0.1", "::1"})
|
||||||
|
|
||||||
|
|
||||||
def _validate_external_endpoint_url(value: SecretStr) -> SecretStr:
|
def _validate_external_endpoint_url(value: SecretStr) -> SecretStr:
|
||||||
"""Valide la structure et le schéma d'une URL d'endpoint externe.
|
"""Valide la structure et le schéma réseau d'une URL d'endpoint externe.
|
||||||
|
|
||||||
|
Le socle commun accepte uniquement les schémas ``http`` et ``https``, avec
|
||||||
|
un hôte obligatoire. Les credentials embarqués (``user:pass@host``) sont
|
||||||
|
refusés afin qu'aucun secret ne soit transporté dans l'URL. La restriction
|
||||||
|
``https``/HTTP loopback est ensuite affinée par chaque connecteur.
|
||||||
|
|
||||||
:param value: URL potentiellement sensible à valider.
|
:param value: URL potentiellement sensible à valider.
|
||||||
:return: URL validée, toujours encapsulée dans ``SecretStr``.
|
:return: URL validée, toujours encapsulée dans ``SecretStr``.
|
||||||
:rtype: SecretStr
|
:rtype: SecretStr
|
||||||
:raises ValueError: Si l'URL est malformée ou utilise un schéma inconnu.
|
:raises ValueError: Si l'URL est malformée, sans hôte, utilise un schéma
|
||||||
|
non réseau ou contient des credentials.
|
||||||
"""
|
"""
|
||||||
is_valid = False
|
is_valid = False
|
||||||
try:
|
try:
|
||||||
@@ -44,8 +50,9 @@ def _validate_external_endpoint_url(value: SecretStr) -> SecretStr:
|
|||||||
_ = parsed.port
|
_ = parsed.port
|
||||||
is_valid = (
|
is_valid = (
|
||||||
parsed.scheme in _EXTERNAL_ENDPOINT_SCHEMES
|
parsed.scheme in _EXTERNAL_ENDPOINT_SCHEMES
|
||||||
and (parsed.scheme not in {"http", "https"} or parsed.hostname is not None)
|
and parsed.hostname is not None
|
||||||
and (parsed.scheme != "file" or bool(parsed.path))
|
and parsed.username is None
|
||||||
|
and parsed.password is None
|
||||||
)
|
)
|
||||||
except ValueError:
|
except ValueError:
|
||||||
pass
|
pass
|
||||||
@@ -60,10 +67,11 @@ EndpointUrl = Annotated[SecretStr, AfterValidator(_validate_external_endpoint_ur
|
|||||||
class ExternalEndpoint(BaseModel):
|
class ExternalEndpoint(BaseModel):
|
||||||
"""Représente un endpoint externe potentiellement sensible.
|
"""Représente un endpoint externe potentiellement sensible.
|
||||||
|
|
||||||
Le socle accepte les transports ``https``, ``http`` et ``file``. Chaque
|
Le socle accepte uniquement les transports réseau ``https`` et ``http``
|
||||||
connecteur restreint ensuite cette liste selon sa propre politique de
|
(hôte obligatoire, sans credentials embarqués). Chaque connecteur restreint
|
||||||
sécurité. L'URL reste encapsulée dans :class:`pydantic.SecretStr` et sa
|
ensuite cette liste selon sa propre politique de sécurité. L'URL reste
|
||||||
sérialisation conserve uniquement une représentation expurgée.
|
encapsulée dans :class:`pydantic.SecretStr` et sa sérialisation conserve
|
||||||
|
uniquement une représentation expurgée.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
model_config = ConfigDict(extra="forbid", frozen=True, hide_input_in_errors=True)
|
model_config = ConfigDict(extra="forbid", frozen=True, hide_input_in_errors=True)
|
||||||
@@ -93,6 +101,7 @@ class PronoteSettings(BaseSettings):
|
|||||||
env_nested_delimiter="__",
|
env_nested_delimiter="__",
|
||||||
extra="ignore",
|
extra="ignore",
|
||||||
env_prefix="PRONOTE_",
|
env_prefix="PRONOTE_",
|
||||||
|
hide_input_in_errors=True,
|
||||||
)
|
)
|
||||||
|
|
||||||
endpoint: ExternalEndpoint | None = None
|
endpoint: ExternalEndpoint | None = None
|
||||||
@@ -153,22 +162,23 @@ class PronoteSettings(BaseSettings):
|
|||||||
def _validate_endpoint_policies(self) -> PronoteSettings:
|
def _validate_endpoint_policies(self) -> PronoteSettings:
|
||||||
"""Applique les transports autorisés aux deux endpoints Pronote.
|
"""Applique les transports autorisés aux deux endpoints Pronote.
|
||||||
|
|
||||||
L'API Pronote utilise HTTPS. Le flux iCal accepte également ``file``
|
L'API Pronote et le flux iCal exigent tous deux HTTPS : aucun fichier
|
||||||
afin de préserver les fixtures locales injectées.
|
local n'est accepté.
|
||||||
|
|
||||||
:return: Instance validée inchangée.
|
:return: Instance validée inchangée.
|
||||||
:rtype: PronoteSettings
|
:rtype: PronoteSettings
|
||||||
:raises ValueError: Si un endpoint utilise un schéma interdit.
|
:raises ValueError: Si un endpoint n'utilise pas HTTPS.
|
||||||
"""
|
"""
|
||||||
if (
|
if (
|
||||||
self.endpoint is not None
|
self.endpoint is not None
|
||||||
and urlparse(self.endpoint.url.get_secret_value()).scheme != "https"
|
and urlparse(self.endpoint.url.get_secret_value()).scheme != "https"
|
||||||
):
|
):
|
||||||
raise ValueError("URL Pronote invalide : HTTPS requis") from None
|
raise ValueError("URL Pronote invalide : HTTPS requis") from None
|
||||||
if self.ical_endpoint is not None and urlparse(
|
if (
|
||||||
self.ical_endpoint.url.get_secret_value()
|
self.ical_endpoint is not None
|
||||||
).scheme not in {"https", "file"}:
|
and urlparse(self.ical_endpoint.url.get_secret_value()).scheme != "https"
|
||||||
raise ValueError("URL iCal Pronote invalide : HTTPS ou file requis") from None
|
):
|
||||||
|
raise ValueError("URL iCal Pronote invalide : HTTPS requis") from None
|
||||||
return self
|
return self
|
||||||
|
|
||||||
@field_serializer("qr_pin")
|
@field_serializer("qr_pin")
|
||||||
@@ -215,6 +225,7 @@ class CalDAVSettings(BaseSettings):
|
|||||||
env_nested_delimiter="__",
|
env_nested_delimiter="__",
|
||||||
extra="ignore",
|
extra="ignore",
|
||||||
env_prefix="CALDAV_",
|
env_prefix="CALDAV_",
|
||||||
|
hide_input_in_errors=True,
|
||||||
)
|
)
|
||||||
|
|
||||||
allow_insecure_http: bool = False
|
allow_insecure_http: bool = False
|
||||||
@@ -233,6 +244,14 @@ class CalDAVSettings(BaseSettings):
|
|||||||
def _migrate_legacy_url(cls, data: object) -> object:
|
def _migrate_legacy_url(cls, data: object) -> object:
|
||||||
"""Migre ``url`` vers l'endpoint commun avec un avertissement.
|
"""Migre ``url`` vers l'endpoint commun avec un avertissement.
|
||||||
|
|
||||||
|
L'alias historique ``CALDAV_URL`` conserve ses sémantiques passées
|
||||||
|
pendant la transition : lorsqu'il embarque des identifiants
|
||||||
|
(``user:pass@hôte``), l'endpoint est construit sans revalidation pour
|
||||||
|
ne pas casser une configuration existante, alors que le contrat
|
||||||
|
canonique ``CALDAV_ENDPOINT__URL`` refuse désormais les credentials
|
||||||
|
embarqués. La politique de transport (HTTPS, ou HTTP loopback
|
||||||
|
uniquement avec ``allow_insecure_http``) reste appliquée ensuite.
|
||||||
|
|
||||||
:param data: Données brutes du modèle.
|
:param data: Données brutes du modèle.
|
||||||
:return: Données complétées avec ``endpoint`` si nécessaire.
|
:return: Données complétées avec ``endpoint`` si nécessaire.
|
||||||
:rtype: object
|
:rtype: object
|
||||||
@@ -246,7 +265,20 @@ class CalDAVSettings(BaseSettings):
|
|||||||
stacklevel=2,
|
stacklevel=2,
|
||||||
)
|
)
|
||||||
if migrated_data.get("endpoint") is None:
|
if migrated_data.get("endpoint") is None:
|
||||||
migrated_data["endpoint"] = {"url": migrated_data["url"]}
|
legacy_url = migrated_data["url"]
|
||||||
|
raw_url = (
|
||||||
|
legacy_url.get_secret_value()
|
||||||
|
if isinstance(legacy_url, SecretStr)
|
||||||
|
else str(legacy_url)
|
||||||
|
)
|
||||||
|
parsed = urlparse(raw_url)
|
||||||
|
if parsed.username is not None or parsed.password is not None:
|
||||||
|
# Alias obsolète : grandfathered, on préserve l'URL telle quelle.
|
||||||
|
migrated_data["endpoint"] = ExternalEndpoint.model_construct(
|
||||||
|
url=legacy_url if isinstance(legacy_url, SecretStr) else SecretStr(legacy_url)
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
migrated_data["endpoint"] = {"url": legacy_url}
|
||||||
return migrated_data
|
return migrated_data
|
||||||
|
|
||||||
@model_validator(mode="after")
|
@model_validator(mode="after")
|
||||||
@@ -412,6 +444,7 @@ class BlogSettings(BaseSettings):
|
|||||||
env_nested_delimiter="__",
|
env_nested_delimiter="__",
|
||||||
extra="ignore",
|
extra="ignore",
|
||||||
env_prefix="BLOG_",
|
env_prefix="BLOG_",
|
||||||
|
hide_input_in_errors=True,
|
||||||
)
|
)
|
||||||
|
|
||||||
enabled: bool = False
|
enabled: bool = False
|
||||||
@@ -449,17 +482,18 @@ class BlogSettings(BaseSettings):
|
|||||||
|
|
||||||
@model_validator(mode="after")
|
@model_validator(mode="after")
|
||||||
def _validate_endpoint_policy(self) -> BlogSettings:
|
def _validate_endpoint_policy(self) -> BlogSettings:
|
||||||
"""Refuse les transports non sûrs pour le flux RSS de production.
|
"""Refuse les transports non sûrs pour le flux RSS.
|
||||||
|
|
||||||
Le transport ``file`` reste autorisé pour les fixtures locales.
|
Seul HTTPS est accepté : aucun fichier local n'est lu depuis un
|
||||||
|
endpoint externe.
|
||||||
|
|
||||||
:return: Instance validée inchangée.
|
:return: Instance validée inchangée.
|
||||||
:rtype: BlogSettings
|
:rtype: BlogSettings
|
||||||
:raises ValueError: Si le schéma n'est ni ``https`` ni ``file``.
|
:raises ValueError: Si le schéma n'est pas ``https``.
|
||||||
"""
|
"""
|
||||||
scheme = urlparse(self.endpoint.url.get_secret_value()).scheme
|
scheme = urlparse(self.endpoint.url.get_secret_value()).scheme
|
||||||
if scheme not in {"https", "file"}:
|
if scheme != "https":
|
||||||
raise ValueError("URL RSS invalide : HTTPS ou file requis") from None
|
raise ValueError("URL RSS invalide : HTTPS requis") from None
|
||||||
return self
|
return self
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -55,14 +55,15 @@ def real_parsed_feed(blog_rss_fixture_path: Path) -> feedparser.FeedParserDict:
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def blog_client(blog_rss_fixture_path: Path) -> BlogRSSClient:
|
def blog_client() -> BlogRSSClient:
|
||||||
"""Instance de BlogRSSClient pointant vers le fixture local.
|
"""Instance de BlogRSSClient pointant vers une URL HTTPS non-réseau.
|
||||||
|
|
||||||
|
Les tests mockent ``requests.get`` : aucune requête réelle n'est émise.
|
||||||
|
|
||||||
:param blog_rss_fixture_path: Chemin vers le fichier fixture.
|
|
||||||
:return: Instance de BlogRSSClient.
|
:return: Instance de BlogRSSClient.
|
||||||
:rtype: BlogRSSClient
|
:rtype: BlogRSSClient
|
||||||
"""
|
"""
|
||||||
return BlogRSSClient(rss_url=f"file://{blog_rss_fixture_path}")
|
return BlogRSSClient(rss_url="https://example.com/blog/feed")
|
||||||
|
|
||||||
|
|
||||||
# --- Helper functions for mocking ---
|
# --- Helper functions for mocking ---
|
||||||
|
|||||||
+240
-7
@@ -7,6 +7,7 @@ textuelles et sérialisées, et que le rechargement fonctionne comme attendu.
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from collections.abc import Callable
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
@@ -143,8 +144,7 @@ def test_url_from_pronote_url_env_var(monkeypatch: MonkeyPatch) -> None:
|
|||||||
"url",
|
"url",
|
||||||
[
|
[
|
||||||
"https://endpoint.example.test/api",
|
"https://endpoint.example.test/api",
|
||||||
"http://localhost:8080/test",
|
"https://endpoint.example.test:8443/ical?icalsecurise=TOKEN", # pragma: allowlist secret
|
||||||
"file:///tmp/fixture.ics",
|
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
def test_external_endpoint_accepts_supported_schemes(url: str) -> None:
|
def test_external_endpoint_accepts_supported_schemes(url: str) -> None:
|
||||||
@@ -177,11 +177,16 @@ def test_external_endpoint_is_immutable_and_redacted() -> None:
|
|||||||
|
|
||||||
:return: None
|
:return: None
|
||||||
"""
|
"""
|
||||||
endpoint = ExternalEndpoint(url=SecretStr("https://user:secret@example.test/calendar"))
|
secret = "SECRET_TOKEN_XYZ" # pragma: allowlist secret
|
||||||
|
endpoint = ExternalEndpoint(
|
||||||
|
url=SecretStr(f"https://example.test/calendar?icalsecurise={secret}")
|
||||||
|
)
|
||||||
with pytest.raises(ValidationError):
|
with pytest.raises(ValidationError):
|
||||||
endpoint.url = SecretStr("https://other.example.test")
|
endpoint.url = SecretStr("https://other.example.test")
|
||||||
assert "secret" not in endpoint.model_dump_json()
|
assert secret not in endpoint.model_dump_json()
|
||||||
assert "REDACTED" in endpoint.model_dump_json()
|
assert "REDACTED" in endpoint.model_dump_json()
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
ExternalEndpoint(url=SecretStr("https://example.test/calendar"), unknown_field="x") # type: ignore[call-arg]
|
||||||
|
|
||||||
|
|
||||||
def test_caldav_endpoint_loads_from_nested_environment(monkeypatch: MonkeyPatch) -> None:
|
def test_caldav_endpoint_loads_from_nested_environment(monkeypatch: MonkeyPatch) -> None:
|
||||||
@@ -214,7 +219,7 @@ def test_pronote_endpoints_load_from_nested_environment(monkeypatch: MonkeyPatch
|
|||||||
:return: None
|
:return: None
|
||||||
"""
|
"""
|
||||||
monkeypatch.setenv("PRONOTE_ENDPOINT__URL", "https://pronote.example.test/parent.html")
|
monkeypatch.setenv("PRONOTE_ENDPOINT__URL", "https://pronote.example.test/parent.html")
|
||||||
monkeypatch.setenv("PRONOTE_ICAL_ENDPOINT__URL", "file:///tmp/pronote.ics")
|
monkeypatch.setenv("PRONOTE_ICAL_ENDPOINT__URL", "https://ical.example.test/pronote.ics")
|
||||||
settings = load_settings()
|
settings = load_settings()
|
||||||
assert settings.pronote.endpoint is not None
|
assert settings.pronote.endpoint is not None
|
||||||
assert settings.pronote.ical_endpoint is not None
|
assert settings.pronote.ical_endpoint is not None
|
||||||
@@ -222,14 +227,17 @@ def test_pronote_endpoints_load_from_nested_environment(monkeypatch: MonkeyPatch
|
|||||||
settings.pronote.endpoint.url.get_secret_value()
|
settings.pronote.endpoint.url.get_secret_value()
|
||||||
== "https://pronote.example.test/parent.html"
|
== "https://pronote.example.test/parent.html"
|
||||||
)
|
)
|
||||||
assert settings.pronote.ical_endpoint.url.get_secret_value() == "file:///tmp/pronote.ics"
|
assert (
|
||||||
|
settings.pronote.ical_endpoint.url.get_secret_value()
|
||||||
|
== "https://ical.example.test/pronote.ics"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
("field", "url", "message"),
|
("field", "url", "message"),
|
||||||
[
|
[
|
||||||
("endpoint", "http://pronote.example.test", "HTTPS requis"),
|
("endpoint", "http://pronote.example.test", "HTTPS requis"),
|
||||||
("ical_endpoint", "http://pronote.example.test/calendar", "HTTPS ou file requis"),
|
("ical_endpoint", "http://pronote.example.test/calendar", "URL iCal Pronote invalide"),
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
def test_pronote_endpoint_policy_rejects_insecure_url(field: str, url: str, message: str) -> None:
|
def test_pronote_endpoint_policy_rejects_insecure_url(field: str, url: str, message: str) -> None:
|
||||||
@@ -492,3 +500,228 @@ def test_sync_future_days_positive_env_loading(monkeypatch: MonkeyPatch) -> None
|
|||||||
monkeypatch.setenv("SYNC_FUTURE_DAYS", "30")
|
monkeypatch.setenv("SYNC_FUTURE_DAYS", "30")
|
||||||
settings = load_settings()
|
settings = load_settings()
|
||||||
assert settings.app.sync_future_days == 30
|
assert settings.app.sync_future_days == 30
|
||||||
|
|
||||||
|
|
||||||
|
# --- Contrat des endpoints externes (matrice partagée) ---
|
||||||
|
|
||||||
|
_ConnectorSettings = CalDAVSettings | PronoteSettings | BlogSettings
|
||||||
|
_CONNECTOR_BUILDERS: dict[str, Callable[[str], _ConnectorSettings]] = {
|
||||||
|
"caldav": lambda url: CalDAVSettings(endpoint=ExternalEndpoint(url=SecretStr(url))),
|
||||||
|
"pronote_ical": lambda url: PronoteSettings(ical_endpoint=ExternalEndpoint(url=SecretStr(url))),
|
||||||
|
"blog": lambda url: BlogSettings(endpoint=ExternalEndpoint(url=SecretStr(url))),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def test_external_endpoint_matrix_accepts_https() -> None:
|
||||||
|
"""HTTPS est accepté par les trois connecteurs (CalDAV, Pronote iCal, Blog RSS).
|
||||||
|
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
url = "https://endpoint.example.test/feed"
|
||||||
|
for name, build in _CONNECTOR_BUILDERS.items():
|
||||||
|
settings = build(url)
|
||||||
|
endpoint = getattr(settings, "endpoint", None) or settings.ical_endpoint # type: ignore[union-attr]
|
||||||
|
assert endpoint is not None
|
||||||
|
assert endpoint.url.get_secret_value() == url, name
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("connector", ["caldav", "pronote_ical", "blog"])
|
||||||
|
def test_external_endpoint_matrix_rejects_file_scheme(connector: str) -> None:
|
||||||
|
"""Le schéma ``file://`` est refusé par les trois connecteurs.
|
||||||
|
|
||||||
|
Contrat corrigé : les endpoints externes doivent être réseau (HTTPS) ;
|
||||||
|
aucun connecteur n'accepte un fichier local.
|
||||||
|
|
||||||
|
:param connector: Nom du connecteur testé.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
_CONNECTOR_BUILDERS[connector]("file:///tmp/fixture.ics")
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
("connector", "url"),
|
||||||
|
[
|
||||||
|
("caldav", "http://endpoint.example.test/feed"),
|
||||||
|
("pronote_ical", "http://pronote.example.test/ical.ics"),
|
||||||
|
("blog", "http://blog.example.test/feed"),
|
||||||
|
("caldav", "ftp://endpoint.example.test/feed"),
|
||||||
|
("pronote_ical", "gopher://pronote.example.test/ical.ics"),
|
||||||
|
("blog", "ftp://blog.example.test/feed"),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_external_endpoint_matrix_rejects_insecure_and_other_schemes(
|
||||||
|
connector: str, url: str
|
||||||
|
) -> None:
|
||||||
|
"""HTTP non-loopback et les schémas non HTTP(S) sont refusés par les trois connecteurs.
|
||||||
|
|
||||||
|
:param connector: Nom du connecteur testé.
|
||||||
|
:param url: URL à refuser.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
_CONNECTOR_BUILDERS[connector](url)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
("connector", "url"),
|
||||||
|
[
|
||||||
|
("caldav", "http://localhost:8080/dav"),
|
||||||
|
("caldav", "http://127.0.0.1:8080/dav"),
|
||||||
|
("caldav", "http://[::1]:8080/dav"),
|
||||||
|
("pronote_ical", "http://localhost/ical.ics"),
|
||||||
|
("blog", "http://localhost/feed"),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_external_endpoint_matrix_rejects_loopback_http_by_default(
|
||||||
|
connector: str, url: str
|
||||||
|
) -> None:
|
||||||
|
"""HTTP loopback est refusé sans autorisation explicite du mode non sécurisé.
|
||||||
|
|
||||||
|
:param connector: Nom du connecteur testé.
|
||||||
|
:param url: URL loopback HTTP à refuser.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
_CONNECTOR_BUILDERS[connector](url)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"url",
|
||||||
|
["http://localhost:8080/dav", "http://127.0.0.1:8080/dav", "http://[::1]:8080/dav"],
|
||||||
|
)
|
||||||
|
def test_caldav_endpoint_accepts_loopback_http_only_with_insecure_flag(url: str) -> None:
|
||||||
|
"""CalDAV accepte HTTP loopback uniquement avec ``allow_insecure_http=True``.
|
||||||
|
|
||||||
|
:param url: URL loopback HTTP à accepter.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
settings = CalDAVSettings(
|
||||||
|
endpoint=ExternalEndpoint(url=SecretStr(url)), allow_insecure_http=True
|
||||||
|
)
|
||||||
|
assert settings.endpoint is not None
|
||||||
|
assert settings.endpoint.url.get_secret_value() == url
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
("connector", "url"),
|
||||||
|
[
|
||||||
|
("caldav", "https:///missing-host"),
|
||||||
|
("pronote_ical", "https:///missing-host"),
|
||||||
|
("blog", "https:///missing-host"),
|
||||||
|
("caldav", "https://host:bad"),
|
||||||
|
("pronote_ical", "https://host:bad"),
|
||||||
|
("blog", "https://host:bad"),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_external_endpoint_matrix_rejects_malformed_urls(connector: str, url: str) -> None:
|
||||||
|
"""Les URL malformées (hôte manquant, port invalide) sont refusées.
|
||||||
|
|
||||||
|
:param connector: Nom du connecteur testé.
|
||||||
|
:param url: URL malformée à refuser.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
_CONNECTOR_BUILDERS[connector](url)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"connector",
|
||||||
|
["caldav", "pronote_ical", "blog"],
|
||||||
|
)
|
||||||
|
def test_external_endpoint_matrix_rejects_userinfo(connector: str) -> None:
|
||||||
|
"""Les credentials dans l'URL (``user:pass@host``) sont refusés.
|
||||||
|
|
||||||
|
:param connector: Nom du connecteur testé.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
url = "https://user:pass@endpoint.example.test/feed" # pragma: allowlist secret
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
_CONNECTOR_BUILDERS[connector](url)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"connector",
|
||||||
|
["caldav", "pronote_ical", "blog"],
|
||||||
|
)
|
||||||
|
def test_external_endpoint_matrix_redacts_sensitive_query_params(connector: str) -> None:
|
||||||
|
"""Les paramètres sensibles de la query sont masqués à la sérialisation.
|
||||||
|
|
||||||
|
:param connector: Nom du connecteur testé.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
secret = "SECRET_QUERY_TOKEN" # pragma: allowlist secret
|
||||||
|
url = f"https://endpoint.example.test/feed?icalsecurise={secret}"
|
||||||
|
settings = _CONNECTOR_BUILDERS[connector](url)
|
||||||
|
endpoint = getattr(settings, "endpoint", None) or settings.ical_endpoint # type: ignore[union-attr]
|
||||||
|
assert endpoint is not None
|
||||||
|
dumped = endpoint.model_dump_json()
|
||||||
|
assert secret not in dumped
|
||||||
|
assert "REDACTED" in dumped
|
||||||
|
assert secret not in repr(endpoint)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
("connector", "legacy_field", "canonical_url", "legacy_url"),
|
||||||
|
[
|
||||||
|
("caldav", "url", "https://canonical.example.test/dav", "https://legacy.example.test/dav"),
|
||||||
|
(
|
||||||
|
"blog",
|
||||||
|
"rss_url",
|
||||||
|
"https://canonical.example.test/feed",
|
||||||
|
"https://legacy.example.test/feed",
|
||||||
|
),
|
||||||
|
(
|
||||||
|
"pronote",
|
||||||
|
"url",
|
||||||
|
"https://canonical.example.test/parent.html",
|
||||||
|
"https://legacy.example.test/parent.html",
|
||||||
|
),
|
||||||
|
(
|
||||||
|
"pronote",
|
||||||
|
"ical_url",
|
||||||
|
"https://canonical.example.test/ical.ics",
|
||||||
|
"https://legacy.example.test/ical.ics",
|
||||||
|
),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_deprecated_aliases_warn_and_canonical_wins(
|
||||||
|
connector: str, legacy_field: str, canonical_url: str, legacy_url: str
|
||||||
|
) -> None:
|
||||||
|
"""Les alias obsolètes émettent un DeprecationWarning et l'URL canonique gagne.
|
||||||
|
|
||||||
|
:param connector: Connecteur testé (``caldav``, ``blog`` ou ``pronote``).
|
||||||
|
:param legacy_field: Nom du champ obsolète.
|
||||||
|
:param canonical_url: URL canonique (doit gagner).
|
||||||
|
:param legacy_url: URL fournie via l'alias obsolète.
|
||||||
|
:return: None
|
||||||
|
"""
|
||||||
|
if connector == "caldav":
|
||||||
|
caldav_kwargs: dict[str, object] = {
|
||||||
|
legacy_field: SecretStr(legacy_url),
|
||||||
|
"endpoint": {"url": canonical_url},
|
||||||
|
}
|
||||||
|
with pytest.warns(DeprecationWarning, match="CALDAV_URL"):
|
||||||
|
caldav_settings = CalDAVSettings(**caldav_kwargs) # type: ignore[arg-type]
|
||||||
|
assert caldav_settings.endpoint is not None
|
||||||
|
assert caldav_settings.endpoint.url.get_secret_value() == canonical_url
|
||||||
|
elif connector == "blog":
|
||||||
|
blog_kwargs: dict[str, object] = {
|
||||||
|
legacy_field: legacy_url,
|
||||||
|
"endpoint": {"url": canonical_url},
|
||||||
|
}
|
||||||
|
with pytest.warns(DeprecationWarning, match="BLOG_RSS_URL"):
|
||||||
|
blog_settings = BlogSettings(**blog_kwargs) # type: ignore[arg-type]
|
||||||
|
assert blog_settings.endpoint.url.get_secret_value() == canonical_url
|
||||||
|
else:
|
||||||
|
canonical_field = "endpoint" if legacy_field == "url" else "ical_endpoint"
|
||||||
|
legacy_value: object = SecretStr(legacy_url) if legacy_field == "ical_url" else legacy_url
|
||||||
|
pronote_kwargs: dict[str, object] = {
|
||||||
|
legacy_field: legacy_value,
|
||||||
|
canonical_field: {"url": canonical_url},
|
||||||
|
}
|
||||||
|
with pytest.warns(DeprecationWarning, match="PRONOTE_"):
|
||||||
|
pronote_settings = PronoteSettings(**pronote_kwargs) # type: ignore[arg-type]
|
||||||
|
endpoint = getattr(pronote_settings, canonical_field)
|
||||||
|
assert endpoint is not None
|
||||||
|
assert endpoint.url.get_secret_value() == canonical_url
|
||||||
|
|||||||
@@ -54,7 +54,7 @@ def fixture_mock_settings() -> Settings:
|
|||||||
return Settings(
|
return Settings(
|
||||||
pronote=PronoteSettings(
|
pronote=PronoteSettings(
|
||||||
url="https://pronote.example.com",
|
url="https://pronote.example.com",
|
||||||
ical_url=SecretStr("file:///fake/ical.ics"),
|
ical_url=SecretStr("https://ical.example.test/ical.ics"),
|
||||||
agenda_source="auto",
|
agenda_source="auto",
|
||||||
homework_source="auto",
|
homework_source="auto",
|
||||||
username="testuser",
|
username="testuser",
|
||||||
@@ -287,7 +287,8 @@ def test_fetch_agenda_ical_mode_failure(mock_fetcher: PronoteFetcher) -> None:
|
|||||||
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
):
|
):
|
||||||
m_fetch_ical.side_effect = OSError(
|
m_fetch_ical.side_effect = OSError(
|
||||||
"Impossible de lire le fichier iCal file:///fake/ical.ics : iCal unreachable"
|
"Échec de la récupération du flux iCal https://ical.example.test/ical.ics : "
|
||||||
|
"iCal unreachable"
|
||||||
)
|
)
|
||||||
m_parse_ical.side_effect = OSError("iCal parse error")
|
m_parse_ical.side_effect = OSError("iCal parse error")
|
||||||
|
|
||||||
@@ -296,7 +297,7 @@ def test_fetch_agenda_ical_mode_failure(mock_fetcher: PronoteFetcher) -> None:
|
|||||||
|
|
||||||
assert "Impossible de récupérer l'agenda : la source ical a échoué" in str(exc_info.value)
|
assert "Impossible de récupérer l'agenda : la source ical a échoué" in str(exc_info.value)
|
||||||
# Vérifie que le message ne contient pas de secret
|
# Vérifie que le message ne contient pas de secret
|
||||||
assert "file:///fake/ical.ics" not in str(exc_info.value)
|
assert "https://ical.example.test/ical.ics" not in str(exc_info.value)
|
||||||
|
|
||||||
|
|
||||||
def test_fetch_agenda_pronotepy_mode_failure(mock_fetcher: PronoteFetcher) -> None:
|
def test_fetch_agenda_pronotepy_mode_failure(mock_fetcher: PronoteFetcher) -> None:
|
||||||
@@ -538,8 +539,8 @@ def test_no_secrets_in_error_messages(
|
|||||||
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
|
||||||
):
|
):
|
||||||
error_msg = (
|
error_msg = (
|
||||||
"Impossible de lire le fichier iCal file:///ical?icalsecurise=SECRET_TOKEN_123 : "
|
"Échec de la récupération du flux iCal https://ical.example.test/ical.ics"
|
||||||
"[Errno 2] No such file or directory"
|
"?icalsecurise=SECRET_TOKEN_123 : [Errno 2] No such file or directory"
|
||||||
)
|
)
|
||||||
m_fetch_ical.side_effect = OSError(error_msg)
|
m_fetch_ical.side_effect = OSError(error_msg)
|
||||||
m_parse_ical.side_effect = OSError("parse error")
|
m_parse_ical.side_effect = OSError("parse error")
|
||||||
|
|||||||
+22
-23
@@ -1,7 +1,7 @@
|
|||||||
"""Tests unitaires pour le module iCal : téléchargement et parsing.
|
"""Tests unitaires pour le module iCal : téléchargement et parsing.
|
||||||
|
|
||||||
Ce module teste :
|
Ce module teste :
|
||||||
- La récupération du flux iCal (file://, HTTP)
|
- La récupération du flux iCal (HTTP)
|
||||||
- Le parsing des événements en modèles Lesson, SchoolEvent
|
- Le parsing des événements en modèles Lesson, SchoolEvent
|
||||||
- L'extraction et normalisation des devoirs
|
- L'extraction et normalisation des devoirs
|
||||||
- La collecte et déduplication des devoirs par date cible
|
- La collecte et déduplication des devoirs par date cible
|
||||||
@@ -11,8 +11,6 @@ Les tests utilisent des mocks pour éviter tout appel réseau réel.
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import tempfile
|
|
||||||
import urllib.parse
|
|
||||||
from datetime import date, datetime
|
from datetime import date, datetime
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
@@ -57,38 +55,39 @@ def invalid_ical_content() -> str:
|
|||||||
return "INVALID:CONTENT\nThis is not a valid iCal file."
|
return "INVALID:CONTENT\nThis is not a valid iCal file."
|
||||||
|
|
||||||
|
|
||||||
def test_fetch_ical_file_protocol() -> None:
|
@responses.activate
|
||||||
"""fetch_ical("file://tests/fixtures/pronote-4e.ics") retourne un contenu commençant par BEGIN:VCALENDAR.
|
def test_fetch_ical_fixture_content() -> None:
|
||||||
|
"""fetch_ical("https://…") sur un mock réseau retourne un contenu commençant par BEGIN:VCALENDAR.
|
||||||
|
|
||||||
:return: None
|
:return: None
|
||||||
"""
|
"""
|
||||||
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
|
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
|
||||||
url = f"file://{fixture_path}"
|
responses.add(
|
||||||
|
responses.GET,
|
||||||
|
"https://pronote.example.test/ical.ics",
|
||||||
|
body=fixture_path.read_text(encoding="utf-8"),
|
||||||
|
status=200,
|
||||||
|
)
|
||||||
|
|
||||||
content = fetch_ical(url)
|
content = fetch_ical("https://pronote.example.test/ical.ics")
|
||||||
assert content.lstrip().startswith("BEGIN:VCALENDAR")
|
assert content.lstrip().startswith("BEGIN:VCALENDAR")
|
||||||
|
|
||||||
|
|
||||||
def test_fetch_ical_file_uri_decoding() -> None:
|
@responses.activate
|
||||||
"""fetch_ical("file://path%20with%20spaces") décode correctement le chemin.
|
def test_fetch_ical_url_with_encoded_characters() -> None:
|
||||||
|
"""fetch_ical préserve les caractères encodés (%20) de l'URL interrogée.
|
||||||
|
|
||||||
:return: None
|
:return: None
|
||||||
"""
|
"""
|
||||||
# Créer un fichier temporaire avec un espace dans le nom
|
responses.add(
|
||||||
with tempfile.TemporaryDirectory() as tmpdir:
|
responses.GET,
|
||||||
temp_path = Path(tmpdir) / "fichier avec espaces.ics"
|
"https://pronote.example.test/fichier%20avec%20espaces.ics",
|
||||||
temp_path.write_text(
|
body="BEGIN:VCALENDAR\nVERSION:2.0\nPRODID:-//Test//Test//FR\nEND:VCALENDAR",
|
||||||
"BEGIN:VCALENDAR\nVERSION:2.0\nPRODID:-//Test//Test//FR\nEND:VCALENDAR",
|
status=200,
|
||||||
encoding="utf-8",
|
)
|
||||||
)
|
|
||||||
|
|
||||||
# URL encodée avec espace
|
content = fetch_ical("https://pronote.example.test/fichier%20avec%20espaces.ics")
|
||||||
encoded_name = urllib.parse.quote("fichier avec espaces.ics")
|
assert content.lstrip().startswith("BEGIN:VCALENDAR")
|
||||||
url = f"file://{tmpdir}/{encoded_name}"
|
|
||||||
|
|
||||||
# Cela devrait fonctionner car Path.read_text décode l'URL
|
|
||||||
content = fetch_ical(url)
|
|
||||||
assert content.lstrip().startswith("BEGIN:VCALENDAR")
|
|
||||||
|
|
||||||
|
|
||||||
@responses.activate
|
@responses.activate
|
||||||
|
|||||||
Reference in New Issue
Block a user