AntoineVe/college-infos
1
0
Fork 0
Code Issues 4 Pull Requests 2 Actions Packages Projects Releases 3 Wiki Activity

Merge remote-tracking branch 'origin/main' into feat/issue-8-account-pin

Browse Source 
# Conflicts:
#	docs/exploitation.md
This commit is contained in:
OpenCode
2026-09-12 23:57:06 +02:00
parent fba5a060df 9433a69d70
commit 52c791695c
11 changed files with 348 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/e2e/test_cli.py
+23 -3
View File
@@ -52,10 +52,10 @@ def test_main_runs_composition_root_in_dry_run_with_requested_log_level(
runner.run.assert_called_once_with()
def test_main_preserves_configured_dry_run_and_returns_success_with_warnings(
def test_main_preserves_configured_dry_run_and_returns_degraded_with_warnings(
mocker: MockerFixture,
) -> None:
"""Sans option, la CLI préserve le dry-run configuré et accepte les avertissements."""
"""Sans option, la CLI préserve le dry-run configuré et signale l'état dégradé."""
from pronote_sync.cli.main import main
settings = Settings(app=AppSettings(dry_run=True, log_level="WARNING"))
@@ -72,12 +72,32 @@ def test_main_preserves_configured_dry_run_and_returns_success_with_warnings(
exit_code = main([])
assert exit_code == 0
assert exit_code == 2
assert setup_logging.call_args_list == [mocker.call("INFO"), mocker.call("WARNING")]
composition_root.assert_called_once_with(settings, dry_run=None)
runner.run.assert_called_once_with()
@pytest.mark.parametrize("step", ["caldav_sync", "send"])
def test_main_returns_degraded_code_for_caldav_or_xmpp_failure(
mocker: MockerFixture,
step: str,
) -> None:
"""Les échecs récupérables CalDAV et XMPP sont observables par le code 2."""
from pronote_sync.cli.main import main
settings = Settings()
mocker.patch("pronote_sync.cli.main.load_settings", return_value=settings)
runner = mocker.Mock()
runner.run.return_value = (
mocker.Mock(spec=PronoteData),
[PipelineWarning(f"Échec récupérable de {step}", step=step)],
)
mocker.patch("pronote_sync.cli.main.PipelineRunner.from_settings", return_value=runner)
assert main([]) == 2
def test_main_returns_failure_and_redacts_pipeline_secrets_at_debug_level(
mocker: MockerFixture,
capsys: pytest.CaptureFixture[str],
tests/integration/test_caldav_sync.py
+85 -5
View File
@@ -7,7 +7,7 @@ ajouts/mises à jour/suppressions, et la préservation des événements non gér
from __future__ import annotations
from datetime import datetime
from datetime import date, datetime
from typing import TYPE_CHECKING, Any
import pytest
@@ -122,10 +122,16 @@ class FakeCalendar:
event_start = raw_start.dt
event_end = raw_end.dt if raw_end is not None else event_start
overlaps = True
if start is not None:
overlaps = overlaps and event_end > start
if end is not None:
overlaps = overlaps and event_start < end
if isinstance(event_start, datetime):
if start is not None:
overlaps = overlaps and event_end > start
if end is not None:
overlaps = overlaps and event_start < end
else:
if start is not None:
overlaps = overlaps and event_end > start.date()
if end is not None:
overlaps = overlaps and event_start < end.date()
if overlaps:
results.append(FakeCalendarEvent(ical_text, uid=uid, server=self._server))
break
@@ -686,6 +692,80 @@ class TestCalDAVSynchronize:
assert result.added == 1
assert len(fake_caldav_server.get_events()) == 1
def test_legacy_homework_vevent_is_migrated_idempotently_and_removed(
self,
fake_caldav_server: FakeCalDAVServer,
full_settings: Settings,
) -> None:
"""Migre un ancien devoir puis vérifie l'idempotence et la suppression."""
fake_caldav_server._events["homework-hw-legacy"] = _create_vevent_text(
uid="homework-hw-legacy",
summary="Devoir: Histoire",
start=datetime(2026, 1, 20, 8, 0),
end=datetime(2026, 1, 20, 18, 0),
status="NEEDS-ACTION",
managed=True,
)
homework = Homework(
id="hw-legacy",
subject="Histoire",
assigned_on=None,
due_on=date(2026, 1, 20),
text="Lire le chapitre 5",
)
def make_data(homeworks: list[Homework]) -> PronoteData:
"""Construit les données de synchronisation du scénario."""
return PronoteData(
lessons=[],
homeworks=homeworks,
school_events=[],
messages=[],
target_date=date(2026, 1, 20),
generated_at=datetime(2026, 1, 14, 0, 0),
)
first = synchronize(
pronote_data=make_data([homework]),
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
assert first.status == CalDAVSyncStatus.SUCCESS
assert first.added == 0
assert first.updated == 1
assert first.removed == 0
migrated = Calendar.from_ical(fake_caldav_server.get_events()["homework-hw-legacy"])
event = migrated.walk("VEVENT")[0]
assert event.get("DTSTART").dt == date(2026, 1, 20)
assert event.get("DTEND").dt == date(2026, 1, 21)
assert event.get("STATUS") is None
assert str(event.get("TRANSP")) == "TRANSPARENT"
second = synchronize(
pronote_data=make_data([homework]),
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
assert second.status == CalDAVSyncStatus.SKIPPED
assert second.added == 0
assert second.updated == 0
assert second.removed == 0
removed = synchronize(
pronote_data=make_data([]),
settings=full_settings,
client_factory=fake_caldav_server.client_factory,
now=datetime(2026, 1, 14, 12, 0),
)
assert removed.status == CalDAVSyncStatus.SUCCESS
assert removed.added == 0
assert removed.updated == 0
assert removed.removed == 1
assert fake_caldav_server.get_events() == {}
def test_school_event_sync(
self,
fake_caldav_server: FakeCalDAVServer,
tests/integration/test_pipeline_runner.py
+104
View File
@@ -1040,6 +1040,110 @@ def test_runner_blog_success_delivers_articles_into_xmpp_message_external_info(
assert xmpp_message.external_info.blog_articles[0].title == "Test Article"
@pytest.mark.parametrize(
("channel_kind", "dry_run", "should_acknowledge"),
[
("success", False, True),
("false", False, False),
("exception", False, False),
("none", False, False),
("success", True, False),
],
)
def test_runner_acknowledges_blog_only_after_confirmed_xmpp_delivery(
pipeline_inputs: tuple[Lesson, Homework],
tmp_path: Any,
channel_kind: str,
dry_run: bool,
should_acknowledge: bool,
) -> None:
"""Les GUID RSS restent rejouables tant que XMPP n'a pas confirmé l'envoi.
:param pipeline_inputs: Données Pronote de test.
:param tmp_path: Répertoire temporaire pour l'état RSS.
:param channel_kind: Comportement du canal XMPP simulé.
:param dry_run: Active ou non le mode simulation.
:param should_acknowledge: Indique si l'état RSS doit être acquitté.
"""
lesson, homework = pipeline_inputs
calls: list[str] = []
state_file = tmp_path / "blog-state.json"
class SuccessfulBlogClient:
"""Client RSS renvoyant un article non encore livré."""
def fetch_and_parse(
self,
*,
known_guids: frozenset[str] | None = None,
etag: str | None = None,
last_modified: str | None = None,
) -> BlogRSSFetchResult:
"""Retourne un article et des en-têtes de cache déterministes.
:param known_guids: GUID déjà connus, ignorés dans ce faux client.
:param etag: ETag mémorisé, ignoré dans ce faux client.
:param last_modified: Date HTTP mémorisée, ignorée dans ce faux client.
:return: Résultat RSS avec un article à livrer.
:rtype: BlogRSSFetchResult
"""
del known_guids, etag, last_modified
return BlogRSSFetchResult(
articles=(
BlogArticle(
id="article-to-deliver",
title="Article à livrer",
url="https://example.com/article-to-deliver",
published_at=datetime(2026, 9, 8, 12, 0),
updated_at=None,
category=None,
author=None,
content_html="<p>Contenu</p>",
content_text="Contenu",
),
),
etag="etag-after-delivery",
last_modified="Tue, 08 Sep 2026 12:00:00 GMT",
)
channel: Any
if channel_kind == "success":
channel = StubChannel(calls)
elif channel_kind == "false":
channel = FailingChannel()
elif channel_kind == "exception":
channel = ExceptionalChannel()
else:
channel = None
runner = PipelineRunner(
settings=Settings(blog=Settings().blog.model_copy(update={"enabled": True})),
pronote_fetcher=StubFetcher(calls, lesson, homework),
caldav_synchronizer=lambda data, settings: successful_sync_result(),
agenda_comparator=cast("AgendaComparator | None", StubComparator(calls)),
blog_client=cast("BlogRSSClient | None", SuccessfulBlogClient()),
blog_state=BlogRSSState(state_file),
channel=channel,
dry_run=dry_run,
now_provider=lambda: datetime(2026, 9, 8, 7, 0),
)
data, errors = runner.run()
assert data is not None
if should_acknowledge:
acknowledged_state = BlogRSSState(state_file)
assert acknowledged_state.get_known_guids() == frozenset({"article-to-deliver"})
assert acknowledged_state.get_cache_headers() == (
"etag-after-delivery",
"Tue, 08 Sep 2026 12:00:00 GMT",
)
else:
assert not state_file.exists()
if channel_kind in {"false", "exception"}:
assert any(error.step == "send" for error in errors)
def test_runner_secret_redaction_in_pipeline_errors(
pipeline_inputs: tuple[Lesson, Homework],
) -> None:
tests/unit/test_blog_state.py
+35
View File
@@ -14,11 +14,14 @@ Tous les tests utilisent des fichiers temporaires via la fixture ``tmp_path``.
from __future__ import annotations
import json
from datetime import UTC, datetime
from pathlib import Path
from unittest.mock import patch
import pytest
from pronote_sync.models.blog import BlogArticle
from pronote_sync.sources.blog.result import BlogRSSFetchResult
from pronote_sync.sources.blog.state import BlogRSSState
@@ -116,6 +119,38 @@ def test_add_guids_empty_noop(tmp_path: Path) -> None:
assert state_file.read_text(encoding="utf-8") == original_content
def test_acknowledge_persists_guids_and_cache_headers_together(tmp_path: Path) -> None:
"""Vérifie l'acquittement atomique après une livraison confirmée.
:param tmp_path: Fixture pytest pour un répertoire temporaire.
:return: None
"""
state_file = tmp_path / "state.json"
state = BlogRSSState(state_file)
article = BlogArticle(
id="guid-1",
title="Article",
url="https://example.com/article",
published_at=datetime(2026, 9, 12, 8, 0, tzinfo=UTC),
updated_at=None,
category=None,
author=None,
content_html="<p>Contenu</p>",
content_text="Contenu",
)
state.acknowledge(
BlogRSSFetchResult(
articles=(article,),
etag="etag-1",
last_modified="Sat, 12 Sep 2026 08:00:00 GMT",
)
)
assert state.get_known_guids() == frozenset({"guid-1"})
assert state.get_cache_headers() == ("etag-1", "Sat, 12 Sep 2026 08:00:00 GMT")
def test_state_load_persisted_guids(tmp_path: Path) -> None:
"""Vérifie que les GUID persistés sont rechargés dans une nouvelle instance.
tests/unit/test_sync_serialization.py
+7 -6
View File
@@ -239,15 +239,16 @@ def test_homework_to_vevent_uid_prefix() -> None:
assert str(event.get("UID")) == "homework-HW-5678"
def test_homework_to_vevent_status() -> None:
"""Vérifie qu'un devoir a STATUS=NEEDS-ACTION.
def test_homework_to_vevent_is_transparent_without_task_status() -> None:
"""Vérifie qu'un devoir VEVENT est transparent et sans statut VTODO.
:return: None
"""
homework = _make_homework()
event = homework_to_vevent(homework)
assert str(event.get("STATUS")) == "NEEDS-ACTION"
assert event.get("STATUS") is None
assert str(event.get("TRANSP")) == "TRANSPARENT"
def test_homework_to_vevent_categories() -> None:
@@ -265,15 +266,15 @@ def test_homework_to_vevent_categories() -> None:
def test_homework_to_vevent_dtstart_dtend() -> None:
"""Vérifie que DTSTART et DTEND couvrent la journée d'échéance (08:00-18:00).
"""Vérifie que DTSTART et DTEND encadrent la journée d'échéance.
:return: None
"""
homework = _make_homework(due_on=date(2026, 1, 20))
event = homework_to_vevent(homework)
assert event.get("DTSTART").dt == datetime(2026, 1, 20, 8, 0)
assert event.get("DTEND").dt == datetime(2026, 1, 20, 18, 0)
assert event.get("DTSTART").dt == date(2026, 1, 20)
assert event.get("DTEND").dt == date(2026, 1, 21)
def test_homework_to_vevent_summary() -> None: