From 30d68a17d3461714ccfe3e0d004d44aae83923a7 Mon Sep 17 00:00:00 2001 From: Michal Medvecky <12493312+michalmedvecky@users.noreply.github.com> Date: Thu, 22 Oct 2026 09:00:31 +0200 Subject: [PATCH 1/2] Add participation APIs and preserve standalone grades --- README.md | 97 ++++++++ edupage_api/__init__.py | 35 +++ edupage_api/consultation_booking.py | 218 ++++++++++++++++ edupage_api/exceptions.py | 32 +++ edupage_api/grades.py | 92 +++++-- edupage_api/registrations.py | 371 ++++++++++++++++++++++++++++ tests/test_consultation_booking.py | 205 +++++++++++++++ tests/test_grades.py | 161 +++++++++++- tests/test_registrations.py | 259 +++++++++++++++++++ 9 files changed, 1441 insertions(+), 29 deletions(-) create mode 100644 edupage_api/consultation_booking.py create mode 100644 edupage_api/registrations.py create mode 100644 tests/test_consultation_booking.py create mode 100644 tests/test_registrations.py diff --git a/README.md b/README.md index 99b2bc1..92b67d5 100644 --- a/README.md +++ b/README.md @@ -42,3 +42,100 @@ The docs are available [here](https://edupageapi.github.io/edupage-api/) # Discord https://discord.gg/fg6zBu9ZAn + +## Registrations, surveys and consultations (read only) + +The following methods use the authenticated session in its current account/child +context. For a parent account, select the intended child with `switch_to_child` +and restore the parent with `switch_to_parent` in a `finally` block. + +```python +# School timezone; change for schools outside this timezone. +actions = edupage.get_registrations(timezone_name="Europe/Prague") +for action in actions: + print(action.name, action.closes_at, action.votable) + # Survey questions retain their EduPage JSON content; answers belong only + # to the current participant. No question JavaScript is executed. + print(action.targets, action.selection) + +history = edupage.get_registration_history() +overview = edupage.get_consultations() +for consultation in overview.consultations: + slots = edupage.get_consultation_slots(consultation.consultation_id) + print(consultation.teacher_name, slots) +print(overview.reservations) +``` + +Assigned actions can include closed registrations. `active`, `votable` and the +opening/closing dates must be considered together. A timeline notification's +completion flag does not establish that a participant answered a survey. +Selection statuses preserve EduPage's values (`ok`, `reserved`, `answered`, or +other values). Parent confirmation and temporary reservation expiry are separate. + +Consultation slots preserve server availability rather than inferring it from +the block duration. Their status is `free`, `reserved`, `temporary`, or `unknown`; +`own` and `closed` are separate flags. Unknown states must never be considered +available. Local timestamps are timezone aware; slot epoch timestamps use UTC. +Different consultation blocks may overlap, so preserve the block identifier. + +These methods do not reserve, cancel, confirm, or submit answers. They send POST +requests for read operations because that is how EduPage loads these views. +HTTP failures, expired login pages and missing expected data raise exceptions; +they do not silently become empty lists. + +Compatibility: HTML/bootstrap and slot shapes were inspected on one parent +account. Tests use synthetic, anonymized examples. Nonempty own reservations, +occupied-slot variants and actual survey question payloads still need live +compatibility testing. Raw survey content is intentionally not interpreted as a +universal questionnaire schema. No credentials or real school data are fixtures. + +### Explicit consultation booking + +Only the methods below perform writes. `dry_run=True` performs the same fresh +availability/ownership checks but never calls reserve, confirm or cancel. + +```python +from datetime import datetime + +plan = edupage.book_consultation_slot( + "3", datetime.fromisoformat("2026-11-10T13:30:00+01:00"), dry_run=True +) +# Only after the user chooses to make this real reservation: +reservation = edupage.book_consultation_slot( + "3", datetime.fromisoformat("2026-11-10T13:30:00+01:00") +) +# Explicit cancellation; verifies ownership and refuses ambiguous blocks: +edupage.cancel_consultation(reservation.visit_id, dry_run=True) +``` + +Booking verifies the server slot, creates a temporary hold, confirms its visit ID +and reads back a confirmed own slot. No automatic retry is made after a write: +`ConsultationUncertain` means the user must inspect EduPage before trying again. +The server's cancellation command addresses an entire consultation block, so the +client first verifies that the requested visit is the only own visit in it. +`ConsultationSlotTaken`, `ConsultationClosed`, `ConsultationLimitReached`, +`ConsultationNotOwned` and `ExpiredSessionException` distinguish known failures. +Unrecognized localized server errors remain a generic `ConsultationError` rather +than being guessed. Participant notes are not supported by the inspected write +protocol; passing a nonempty `note` raises an explicit error before any write. + +Consultation blocks also expose `teaches_student` from EduPage's `coUci` map +(`None` means unknown), and `is_class_teacher` for client enrichment. + +Write behavior is covered by synthetic tests. No real booking or cancellation +was used during development. Real success/error responses still need supervised +validation on a term the user actually intends to book. + +### Standalone grades and read completeness + +Some schools return marks with udalostid=0, without a paper/event. get_grades() +and term reads retain these marks using the row's predmetid, datum, data and +ucitelid. EduGrade.event_id is None for such marks; grade_id retains znamkaid +when provided. Consumers must not confuse these IDs with timeline event IDs. +Missing standalone weights use 1. Existing event metadata continues to supply +weights for marks attached to papers/events. + +The latest successful get_grades() call exposes grade_read_report on Edupage: +received, parsed, skipped and skip_reasons. Copy it before another child/term +read if needed. Skipped rows log field names and reasons without grade values. +Invalid event metadata fails explicitly instead of silently dropping a mark. diff --git a/edupage_api/__init__.py b/edupage_api/__init__.py index f713317..4eae0a7 100644 --- a/edupage_api/__init__.py +++ b/edupage_api/__init__.py @@ -25,6 +25,9 @@ EduTeacher, People, ) +from edupage_api.registrations import ( + Registrations, Registration, RegistrationHistory, ConsultationOverview, ConsultationSlot, +) from edupage_api.ringing import RingingTime, RingingTimes from edupage_api.subjects import Subject, Subjects from edupage_api.substitution import Substitution, TimetableChange @@ -33,6 +36,38 @@ class Edupage(EdupageModule): + def book_consultation_slot(self, consultation_id, starts_at, *, dry_run=False, note=None, + timezone_name="Europe/Prague"): + """Explicitly book and verify a confirmed consultation; dry_run only reads.""" + from edupage_api.consultation_booking import ConsultationBooking + return ConsultationBooking(self, timezone_name).book_consultation_slot( + consultation_id, starts_at, dry_run=dry_run, note=note) + + def cancel_consultation(self, visit_id, *, dry_run=False, timezone_name="Europe/Prague"): + """Cancel only an unambiguously owned visit; dry_run only reads.""" + from edupage_api.consultation_booking import ConsultationBooking + return ConsultationBooking(self, timezone_name).cancel_consultation(visit_id, dry_run=dry_run) + + def get_registrations(self, timezone_name="Europe/Prague") -> list[Registration]: + """Read assigned registration actions and surveys in the current context.""" + return Registrations(self, timezone_name).get_registrations() + + def get_registration(self, action_id: str, timezone_name="Europe/Prague") -> Registration: + """Read an action, its questions and own selections.""" + return Registrations(self, timezone_name).get_registration(action_id) + + def get_registration_history(self, timezone_name="Europe/Prague") -> list[RegistrationHistory]: + """Read own registration participation history.""" + return Registrations(self, timezone_name).get_registration_history() + + def get_consultations(self, timezone_name="Europe/Prague") -> ConsultationOverview: + """Read teacher consultation blocks and own reservations.""" + return Registrations(self, timezone_name).get_consultations() + + def get_consultation_slots(self, consultation_id: str, timezone_name="Europe/Prague") -> list[ConsultationSlot]: + """Read a consultation's slot availability without reserving.""" + return Registrations(self, timezone_name).get_consultation_slots(consultation_id) + def __init__(self, request_timeout=5): """Initialize `Edupage` object. diff --git a/edupage_api/consultation_booking.py b/edupage_api/consultation_booking.py new file mode 100644 index 0000000..5809e90 --- /dev/null +++ b/edupage_api/consultation_booking.py @@ -0,0 +1,218 @@ +"""Explicit consultation writes. Read-only Registrations never calls this module.""" + +import json +import re +from datetime import datetime, timezone +from urllib.parse import parse_qs, urlencode, urlsplit + +from requests import RequestException + +from edupage_api.exceptions import ( + ConsultationClosed, + ConsultationError, + ConsultationLimitReached, + ConsultationNotOwned, + ConsultationSlotNotFound, + ConsultationSlotTaken, + ConsultationUncertain, + ExpiredSessionException, +) +from edupage_api.module import ModuleHelper +from edupage_api.registrations import _HTML, ConsultationReservation, Registrations + + +class ConsultationBooking(Registrations): + def _detail(self, consultation_id): + path = "/calendar/?" + urlencode( + { + "cmd": "KonzultacneHodinyVisit", + "konzultaciaids": str(consultation_id), + "asDialog": "1", + } + ) + text = self._request(path).text + # Only extract the literal transport URL. Never execute a script. + for script in _HTML(text).scripts: + match = re.search(r"\$j\.post\('((?:\\.|[^'\\])*)&akcia=rezervuj'", script) + if match: + literal = json.loads('"' + match[1].replace('"', '\\"') + '"') + parts = urlsplit(literal) + params = parse_qs(parts.query) + if ( + parts.scheme + or parts.netloc + or parts.path != "/calendar/" + or params.get("cmd") != ["KonzultacneHodinyVisit"] + or not params.get("jwid") + ): + raise ConsultationError("Unexpected consultation form") + path = "/calendar/?" + urlencode( + { + "cmd": "KonzultacneHodinyVisit", + "jwid": params["jwid"][0], + "asDialog": "1", + } + ) + return path, self._slots(text, consultation_id) + raise ConsultationError("Consultation booking form missing") + + def _write(self, path, action, data, *, json_required=True): + if action not in ("rezervuj", "confirm", "zrus"): + raise ValueError("Unsupported consultation write") + try: + response = self.edupage.session.post( + f"https://{self.edupage.subdomain}.edupage.org{path}&akcia={action}", + data=data, + ) + except RequestException as exc: + # No retry: the server may already have accepted the operation. + raise ConsultationUncertain( + "Write outcome is unknown; check EduPage" + ) from exc + if "/login" in response.url or re.search( + r"]+type=[\'\"]password", response.text, re.I + ): + raise ExpiredSessionException("EduPage session expired") + if response.status_code != 200: + raise ConsultationUncertain("Write outcome is unknown; check EduPage") + try: + result = response.json() + except ValueError as exc: + if not json_required: + return None # Cancellation is verified by a fresh read. + raise ConsultationUncertain( + "Unexpected write response; check EduPage" + ) from exc + if not isinstance(result, dict): + if not json_required: + return None + raise ConsultationUncertain("Unexpected write response; check EduPage") + if result.get("status") == "ok": + return result + if not json_required and "status" not in result: + return result + # Only recognized machine-readable codes are classified. Do not infer + # server error meanings from localized free text. + code = result.get("code") or (result.get("data") or {}).get("code") + error = { + "slot_taken": ConsultationSlotTaken, + "booking_closed": ConsultationClosed, + "booking_limit": ConsultationLimitReached, + }.get(code, ConsultationError) + raise error("EduPage rejected the consultation request") + + @ModuleHelper.logged_in + def book_consultation_slot( + self, consultation_id, starts_at, *, dry_run=False, note=None + ): + if note: + # The inspected booking protocol has no note field. Never discard it silently. + error = ConsultationError( + "Participant notes are not supported by this protocol" + ) + error.code = "note_unsupported" + raise error + if ( + not isinstance(starts_at, datetime) + or starts_at.tzinfo is None + or starts_at.utcoffset() is None + ): + raise ValueError("starts_at must be a timezone-aware datetime") + now = datetime.now(timezone.utc) + if starts_at <= now: + raise ConsultationClosed("Consultation is in the past") + overview = self.get_consultations() + block = next( + ( + b + for b in overview.consultations + if b.consultation_id == str(consultation_id) + ), + None, + ) + if block is None: + raise ConsultationClosed("Consultation no longer offered") + if (block.booking_opens_at and now < block.booking_opens_at) or ( + block.booking_closes_at and now > block.booking_closes_at + ): + raise ConsultationClosed("Booking window is closed") + path, slots = self._detail(consultation_id) + if any(s.own for s in slots): + raise ConsultationLimitReached( + "An own reservation already exists in this block" + ) + slot = next((s for s in slots if s.starts_at == starts_at), None) + if slot is None: + raise ConsultationSlotNotFound( + "This time does not exist in the consultation block" + ) + if slot.status != "free": + raise ConsultationSlotTaken("Slot is no longer free") + if slot.closed: + raise ConsultationClosed("Slot is closed") + epoch = int(slot.starts_at.timestamp()) + if starts_at.timestamp() != epoch: + raise ValueError("Slot start must match an exact server timeframe") + plan = { + "consultation_id": str(consultation_id), + "start": starts_at.isoformat(), + "reserve": {"konzultaciaid": str(consultation_id), "cas": str(epoch)}, + "confirm": {"visitids": ""}, + } + if dry_run: + return plan + held = self._write(path, "rezervuj", plan["reserve"]) + visit_id = str((held.get("data") or {}).get("visitid") or "") + if not visit_id: + raise ConsultationUncertain( + "Missing visit ID after reservation; check EduPage" + ) + try: + self._write(path, "confirm", {"visitids": visit_id}) + confirmed = self.get_consultation_slots(consultation_id) + except ExpiredSessionException: + raise + except Exception as exc: + raise ConsultationUncertain( + f"Reservation {visit_id} may be held; check EduPage" + ) from exc + if not any( + s.own + and s.visit_id == visit_id + and s.status == "reserved" + and s.starts_at == starts_at + for s in confirmed + ): + raise ConsultationUncertain( + f"Reservation {visit_id} was not verified as confirmed" + ) + return ConsultationReservation(str(consultation_id), starts_at, visit_id) + + @ModuleHelper.logged_in + def cancel_consultation(self, visit_id, *, dry_run=False): + # The server cancels by block, not visit ID. Resolve own ownership first + # and refuse ambiguous blocks, rather than cancelling an unrelated visit. + overview = self.get_consultations() + booking = next( + (r for r in overview.reservations if r.visit_id == str(visit_id)), None + ) + if booking is None: + raise ConsultationNotOwned("Own booking not found") + path, slots = self._detail(booking.consultation_id) + own = [s for s in slots if s.own] + if len(own) != 1 or own[0].visit_id != str(visit_id): + raise ConsultationNotOwned("Booking ownership is ambiguous") + plan = { + "visit_id": str(visit_id), + "cancel": {"konzultaciaid": booking.consultation_id}, + } + if dry_run: + return plan + self._write(path, "zrus", plan["cancel"], json_required=False) + try: + remaining = self.get_consultation_slots(booking.consultation_id) + except Exception as exc: + raise ConsultationUncertain("Cancellation could not be verified") from exc + if any(s.own and s.visit_id == str(visit_id) for s in remaining): + raise ConsultationUncertain("Reservation still present after cancellation") + return None diff --git a/edupage_api/exceptions.py b/edupage_api/exceptions.py index 239b2ed..d6434ec 100644 --- a/edupage_api/exceptions.py +++ b/edupage_api/exceptions.py @@ -82,3 +82,35 @@ class InsufficientPermissionsException(Exception): class CaptchaException(Exception): pass + + +class ConsultationError(Exception): + """A consultation request could not be completed safely.""" + + code = "consultation_rejected" + + +class ConsultationSlotTaken(ConsultationError): + code = "slot_taken" + + +class ConsultationClosed(ConsultationError): + code = "booking_closed" + + +class ConsultationLimitReached(ConsultationError): + code = "booking_limit" + + +class ConsultationNotOwned(ConsultationError): + code = "booking_not_owned" + + +class ConsultationUncertain(ConsultationError): + """A write may have succeeded; inspect the account before retrying.""" + + code = "booking_uncertain" + + +class ConsultationSlotNotFound(ConsultationError): + code = "slot_not_found" diff --git a/edupage_api/grades.py b/edupage_api/grades.py index 03ba0d0..a8d9e66 100644 --- a/edupage_api/grades.py +++ b/edupage_api/grades.py @@ -1,4 +1,5 @@ import json +import logging from dataclasses import dataclass from datetime import datetime from enum import Enum @@ -11,9 +12,12 @@ from edupage_api.people import EduTeacher +_LOGGER = logging.getLogger(__name__) + + @dataclass class EduGrade: - event_id: int + event_id: int | None title: str grade_n: int | float | str | None comment: str | None @@ -27,6 +31,7 @@ class EduGrade: verbal: bool percent: float class_grade_avg: float | None + grade_id: str | None = None @dataclass class EduTextGrade: @@ -118,36 +123,67 @@ def get_grades(self, term: Term | None, year: int | None) -> list[EduGrade]: else self.__get_grade_data() ) + if not isinstance(grade_data, dict): + raise FailedToParseGradeDataError("Grade response must be an object") grades = grade_data.get("vsetkyZnamky") - grade_details = grade_data.get("vsetkyUdalosti").get("edupage") + if not isinstance(grades, list): + raise FailedToParseGradeDataError("Missing/invalid vsetkyZnamky list") + _LOGGER.debug("Grade parser: received=%d term=%s year=%s", len(grades), term, year) + report = {"received": len(grades), "parsed": 0, "skipped": 0, "skip_reasons": {}} + + def skip(row_index, row, reason): + report["skipped"] += 1 + report["skip_reasons"][reason] = report["skip_reasons"].get(reason, 0) + 1 + _LOGGER.warning("Grade row skipped: row=%d reason=%s keys=%s", + row_index, reason, sorted(row)) + + if not grades: + self.edupage.grade_read_report = report + return [] + events = grade_data.get("vsetkyUdalosti") + if not isinstance(events, dict): + raise FailedToParseGradeDataError("Missing/invalid vsetkyUdalosti object") + grade_details = events.get("edupage") + if not isinstance(grade_details, dict): + raise FailedToParseGradeDataError("Missing/invalid vsetkyUdalosti.edupage object") output = [] - for grade in grades: - # ID - event_id_str = grade.get("udalostid") - if not event_id_str: - continue - - event_id = int(event_id_str) - - # Title - details = grade_details.get(event_id_str) - title = details.get("p_meno") - - # Date + for row_index, grade in enumerate(grades): + if not isinstance(grade, dict): + raise FailedToParseGradeDataError("Invalid grade row (expected object)") + # Standalone marks need not refer to a school event/paper. + event_id_raw = grade.get("udalostid") + event_id = int(event_id_raw) if event_id_raw not in (None, "", 0, "0") else None + if event_id is None: + _LOGGER.debug("Grade row without event: row=%d keys=%s", row_index, sorted(grade)) + details = grade + else: + details = grade_details.get(str(event_id)) + if not isinstance(details, dict): + raise FailedToParseGradeDataError( + f"Missing/invalid vsetkyUdalosti.edupage[{event_id}] " + "for an existing grade; refusing to omit it from averages" + ) + title = details.get("p_meno") or "" date_str = grade.get("datum") - date = datetime.strptime(date_str, "%Y-%m-%d %H:%M:%S") - - # Subject ID and name - subject_id_str = details.get("PredmetID") + subject_id_str = details.get("PredmetID", details.get("predmetid")) if subject_id_str is None or subject_id_str == "vsetky": + skip(row_index, grade, "missing_subject_id" if subject_id_str is None else "all_subjects") + continue + if not isinstance(date_str, str) or not isinstance(grade.get("data"), str): + skip(row_index, grade, "missing_date_or_grade") + continue + try: + date = datetime.strptime(date_str, "%Y-%m-%d %H:%M:%S") + except ValueError: + skip(row_index, grade, "invalid_date") continue subject_id = int(subject_id_str) subject_name = DbiHelper(self.edupage).fetch_subject_name(subject_id) # Teacher - teacher_id_str = details.get("UcitelID") + teacher_id_str = details.get("UcitelID", details.get("ucitelid")) if teacher_id_str is None: teacher = None else: @@ -162,11 +198,11 @@ def get_grades(self, term: Term | None, year: int | None) -> list[EduGrade]: # Maximal points and importance grade_type = str(details.get("p_typ_udalosti")) max_points = None - importance = None + importance = 1.0 if event_id is None else None if grade_type == "1": # Normal grade (1 – 5) max_points = None - importance = float(details.get("p_vaha")) / 20 + importance = float(details.get("p_vaha", 20)) / 20 elif grade_type == "2": # Points grade (in points – e.g. 0 – 20 points) max_points = float(details.get("p_vaha")) @@ -174,11 +210,11 @@ def get_grades(self, term: Term | None, year: int | None) -> list[EduGrade]: elif grade_type == "3": # Percental grade (0 – 100 %) max_points = float(details.get("p_vaha_body")) - importance = float(details.get("p_vaha")) / 20 + importance = float(details.get("p_vaha", 20)) / 20 elif details.get("p_vaha") is not None: # Other grade types (e.g. "4" – symbolic grades like "Ú") # have no maximal points, only importance - importance = float(details.get("p_vaha")) / 20 + importance = float(details.get("p_vaha", 20)) / 20 # More details coming from Edupage metadata more_details_raw = details.get("moredata") @@ -230,8 +266,14 @@ def get_grades(self, term: Term | None, year: int | None) -> list[EduGrade]: importance, verbal, percent, - class_grade_avg + class_grade_avg, + str(grade.get("znamkaid", grade.get("ZnamkaID"))) + if grade.get("znamkaid", grade.get("ZnamkaID")) is not None else None, ) output.append(grade) + report["parsed"] += 1 + _LOGGER.debug("Grade parser: received=%d parsed=%d skipped=%d", + len(grades), len(output), len(grades) - len(output)) + self.edupage.grade_read_report = report return output diff --git a/edupage_api/registrations.py b/edupage_api/registrations.py new file mode 100644 index 0000000..3266f46 --- /dev/null +++ b/edupage_api/registrations.py @@ -0,0 +1,371 @@ +"""Read-only registrations, surveys and teacher consultations. + +Uses the existing session and never executes JavaScript or submits a selection. +Question content is exposed as server JSON, since EduPage question types vary. +""" + +import json +import re +from dataclasses import dataclass, field +from datetime import datetime, timezone +from html.parser import HTMLParser +from urllib.parse import parse_qs, urlencode, urlsplit +from zoneinfo import ZoneInfo + +from edupage_api.exceptions import ( + ExpiredSessionException, + MissingDataException, + RequestError, +) +from edupage_api.module import Module, ModuleHelper + + +def _date(value, tz): + return datetime.fromisoformat(value).replace(tzinfo=ZoneInfo(tz)) if value else None + + +def _true(value): + return value is True or value in (1, "1", "yes") + + +class _HTML(HTMLParser): + def __init__(self, text): + super().__init__() + self.scripts = [] + self.slots = [] + self._script = None + self.feed(text) + + def handle_starttag(self, tag, attrs): + attrs = dict(attrs) + if tag == "script" and not attrs.get("src"): + self._script = [] + if "timeframe" in attrs.get("class", "").split() and attrs.get("data-cas"): + self.slots.append(attrs) + + def handle_data(self, data): + if self._script is not None: + self._script.append(data) + + def handle_endtag(self, tag): + if tag == "script" and self._script is not None: + self.scripts.append("".join(self._script)) + self._script = None + + +def _bootstrap(text, widget): + for script in _HTML(text).scripts: + match = re.search(r"\." + re.escape(widget) + r"\s*\(\s*(\{)", script) + if match: + try: + data, _ = json.JSONDecoder().raw_decode(script[match.start(1) :]) + return data + except (ValueError, TypeError) as exc: + raise MissingDataException("Invalid EduPage bootstrap JSON") from exc + raise MissingDataException(f"EduPage bootstrap {widget} missing") + + +@dataclass +class RegistrationSelection: + target_id: str + status: str + answers: object = None + parent_confirmed_at: datetime | None = None + reserved_until: datetime | None = None + + +@dataclass +class RegistrationTarget: + target_id: str + name: str + question: object = None + minimum: int | None = None + maximum: int | None = None + confirmed_count: int | None = None + reserved_count: int | None = None + + +@dataclass +class Registration: + action_id: str + name: str + note: str = "" + opens_at: datetime | None = None + closes_at: datetime | None = None + active: bool = False + votable: bool = False + parent_confirmed: bool = False + targets: list[RegistrationTarget] = field(default_factory=list) + selection: list[RegistrationSelection] = field(default_factory=list) + requires_parent_confirmation: bool = False + + +@dataclass +class RegistrationHistory: + action_id: str + name: str + timestamp: datetime | None + + +@dataclass +class Consultation: + consultation_id: str + teacher_id: str + teacher_name: str + starts_at: datetime | None + ends_at: datetime | None + slot_minutes: int + booking_opens_at: datetime | None + booking_closes_at: datetime | None + note: str = "" + teaches_student: bool | None = None + is_class_teacher: bool = False + + +@dataclass +class ConsultationReservation: + consultation_id: str + starts_at: datetime | None + visit_id: str | None = None + + +@dataclass +class ConsultationOverview: + consultations: list[Consultation] + reservations: list[ConsultationReservation] + + +@dataclass +class ConsultationSlot: + consultation_id: str + starts_at: datetime + status: str + own: bool + visit_id: str | None = None + expires_at: datetime | None = None + closed: bool = False + + +def _number(value): + return int(value) if value not in (None, "") else None + + +class Registrations(Module): + """Read operations in the currently selected account/child context. + + EduPage local timestamps use ``timezone_name`` (default Europe/Prague). + Callers with schools in another timezone must pass that school's timezone. + """ + + def __init__(self, edupage, timezone_name="Europe/Prague"): + super().__init__(edupage) + ZoneInfo(timezone_name) # Validate before any request. + self.timezone = timezone_name + + def _request(self, path, data=None): + parts = urlsplit(path) + if parts.scheme or parts.netloc or parts.path not in ("/signin/", "/calendar/"): + raise ValueError("Unexpected EduPage read endpoint") + query = parse_qs(parts.query, keep_blank_values=True) + if query.get("cmd", [None])[0] not in ( + "SigninHome", + "SigninActionPlayer", + "KonzultacneHodinyScheduler", + "KonzultacneHodinyVisit", + ) or any(k.startswith("ak_") for k in query): + raise ValueError("Unexpected EduPage read command") + if any(action != "getData" for action in query.get("akcia", [])): + raise ValueError("Only read actions are supported") + response = self.edupage.session.post( + f"https://{self.edupage.subdomain}.edupage.org{path}", data=data or {} + ) + if response.status_code != 200: + raise RequestError(f"EduPage returned HTTP {response.status_code}") + if "/login" in response.url or re.search( + r"]+type=[\'\"]password", response.text, re.I + ): + raise ExpiredSessionException("EduPage session expired") + return response + + def _view(self, path, widget, view): + bootstrap = _bootstrap(self._request(path).text, widget) + # Never forward arbitrary URLs/commands from returned JavaScript. + form = urlsplit(bootstrap["formurl"]) + if form.path != urlsplit(path).path or form.scheme or form.netloc: + raise MissingDataException("Unexpected EduPage form URL") + params = parse_qs(form.query) + jwid = params.get("jwid", [None])[0] + if not jwid: + raise MissingDataException("EduPage form identifier missing") + command = parse_qs(urlsplit(path).query)["cmd"][0] + query = urlencode({"cmd": command, "jwid": jwid, "akcia": "getData"}) + response = self._request(f"{form.path}?{query}", {"pohlad": view}) + try: + result = response.json() + except ValueError as exc: + raise MissingDataException("Expected EduPage JSON view") from exc + if not isinstance(result, dict): + raise MissingDataException("Expected EduPage view object") + return result + + @ModuleHelper.logged_in + def get_registrations(self): + """List assigned actions, including closed actions; fetch each detail.""" + data = self._view("/signin/?cmd=SigninHome", "signinHome", "openableActions") + if not isinstance(data.get("actions"), list): + raise MissingDataException("EduPage actions missing") + ids = dict.fromkeys(str(r["actionid"]) for r in data["actions"]) + return [self.get_registration(action_id) for action_id in ids] + + @ModuleHelper.logged_in + def get_registration(self, action_id): + """Read questions, options and own selections without voting.""" + path = "/signin/?" + urlencode( + {"cmd": "SigninActionPlayer", "actionid": str(action_id)} + ) + data = _bootstrap(self._request(path).text, "signinactionplayer") + row = data["actionData"] + targets = [] + for target in row.get("targets") or []: + content = target.get("data") or {} + targets.append( + RegistrationTarget( + str(target["targetid"]), + target.get("name") or "", + content.get("content") if content.get("type") == "card" else None, + _number(target.get("min_users")), + _number(target.get("max_users")), + _number(target.get("confirmed_count")), + _number(target.get("reserved_count")), + ) + ) + selections = [] + for target_id, selection in (data.get("selection") or {}).items(): + answers = selection.get("answers") + if isinstance(answers, str): + answers = json.loads(answers) + selections.append( + RegistrationSelection( + str(target_id), + selection.get("stav") or "unknown", + answers, + _date(selection.get("parent_confirmed"), self.timezone), + _date(selection.get("reserved_till"), self.timezone), + ) + ) + return Registration( + str(row["actionid"]), + row.get("name") or "", + row.get("note") or "", + _date(row.get("time_from"), self.timezone), + _date(row.get("time_to"), self.timezone), + _true(row.get("active")), + _true(data.get("votable")), + _true(data.get("parentConfirmed")), + targets, + selections, + str((row.get("data") or {}).get("parents_block") or "") in ("1", "2", "3"), + ) + + @ModuleHelper.logged_in + def get_registration_history(self): + data = self._view("/signin/?cmd=SigninHome", "signinHome", "history") + if not isinstance(data.get("history"), list): + raise MissingDataException("EduPage history missing") + return [ + RegistrationHistory( + str(r["actionid"]), + r.get("action_name") or "", + _date(r.get("timestamp"), self.timezone), + ) + for r in data["history"] + ] + + @ModuleHelper.logged_in + def get_consultations(self): + """Read offered blocks and reservations of the current account.""" + data = _bootstrap( + self._request("/calendar/?cmd=KonzultacneHodinyScheduler").text, + "konzultacneHodinyScheduler2", + ) + if not isinstance(data.get("hodinyVsetci"), dict): + raise MissingDataException("EduPage consultation blocks missing") + blocks = [] + for rows in data["hodinyVsetci"].values(): + for row in rows: + blocks.append( + Consultation( + str(row["konzultaciaid"]), + str(row["ucitelid"]), + " ".join( + filter(None, (row.get("p_meno"), row.get("p_priezvisko"))) + ), + _date(row.get("cas_od"), self.timezone), + _date(row.get("cas_do"), self.timezone), + int(row["timeframe"]), + _date(row.get("prihlasovanie_odkedy"), self.timezone), + _date(row.get("prihlasovanie_dokedy"), self.timezone), + row.get("poznamka") or "", + bool((data.get("coUci") or {}).get(str(row["ucitelid"]))) + if isinstance(data.get("coUci"), dict) + else None, + ) + ) + if "rezervacie" not in data and "rezervacieMoje" not in data: + raise MissingDataException("EduPage own reservation list missing") + bookings = {} + for row in (data.get("rezervacie") or []) + (data.get("rezervacieMoje") or []): + booking = ConsultationReservation( + str(row["konzultaciaid"]), + _date( + row.get("cas_odkedy") or row.get("cas_rezervacie"), self.timezone + ), + str(row["visitid"]) if row.get("visitid") else None, + ) + if booking.starts_at is None: + raise MissingDataException("EduPage reservation start missing") + bookings[(booking.consultation_id, booking.starts_at)] = booking + return ConsultationOverview(blocks, list(bookings.values())) + + @ModuleHelper.logged_in + def get_consultation_slots(self, consultation_id): + """Read server-rendered slot states. Never click or reserve a slot.""" + path = "/calendar/?" + urlencode( + { + "cmd": "KonzultacneHodinyVisit", + "konzultaciaids": str(consultation_id), + "asDialog": "1", + } + ) + text = self._request(path).text + if "konzultacieVDni" not in text: + raise MissingDataException("EduPage consultation detail missing") + return self._slots(text, consultation_id) + + def _slots(self, text, consultation_id): + result = [] + for row in _HTML(text).slots: + if row.get("data-konzultaciaid") != str(consultation_id): + continue + classes = row.get("class", "").split() + status = ( + "temporary" + if "unconfirmed" in classes + else "reserved" + if "confirmed" in classes + else "free" + if "free" in classes + else "unknown" + ) + result.append( + ConsultationSlot( + str(consultation_id), + datetime.fromtimestamp(int(row["data-cas"]), timezone.utc), + status, + "moje" in classes, + row.get("data-visitid") or None, + _date(row.get("data-platnost"), self.timezone), + "closed" in classes, + ) + ) + return result diff --git a/tests/test_consultation_booking.py b/tests/test_consultation_booking.py new file mode 100644 index 0000000..0496ab7 --- /dev/null +++ b/tests/test_consultation_booking.py @@ -0,0 +1,205 @@ +"""Synthetic write protocol tests; no real school requests.""" + +from datetime import datetime, timedelta, timezone +from unittest.mock import Mock + +import pytest +from edupage_api import Edupage +from edupage_api.consultation_booking import ConsultationBooking +from edupage_api.exceptions import ( + ConsultationClosed, + ConsultationError, + ConsultationLimitReached, + ConsultationNotOwned, + ConsultationSlotTaken, + ConsultationUncertain, + ExpiredSessionException, +) +from edupage_api.registrations import ( + Consultation, + ConsultationOverview, + ConsultationReservation, + ConsultationSlot, +) +from requests import Timeout + + +@pytest.fixture +def booking(): + api = Edupage() + api.subdomain, api.is_logged_in = "school", True + api.session = Mock() + module = ConsultationBooking(api) + future = datetime.now(timezone.utc).replace(microsecond=0) + timedelta(days=3) + block = Consultation( + "b1", + "t1", + "Teacher", + future, + future + timedelta(hours=1), + 30, + None, + future - timedelta(days=1), + teaches_student=True, + ) + module.get_consultations = Mock(return_value=ConsultationOverview([block], [])) + slot = ConsultationSlot("b1", future, "free", False) + module._detail = Mock( + return_value=("/calendar/?cmd=KonzultacneHodinyVisit&jwid=dynamic", [slot]) + ) + module.get_consultation_slots = Mock( + return_value=[ConsultationSlot("b1", future, "reserved", True, "v1")] + ) + api.session.post.side_effect = [ + Mock( + status_code=200, + url="https://school.edupage.org/calendar/", + text="", + json=Mock(return_value={"status": "ok", "data": {"visitid": "v1"}}), + ), + Mock( + status_code=200, + url="https://school.edupage.org/calendar/", + text="", + json=Mock(return_value={"status": "ok"}), + ), + ] + return module, future, block, slot + + +def test_dry_run_performs_no_write(booking): + m, start, _, _ = booking + plan = m.book_consultation_slot("b1", start, dry_run=True) + assert plan["reserve"]["cas"] == str(int(start.timestamp())) + m.edupage.session.post.assert_not_called() + + +def test_book_holds_confirms_and_verifies(booking): + m, start, _, _ = booking + result = m.book_consultation_slot("b1", start) + assert result.visit_id == "v1" + calls = m.edupage.session.post.call_args_list + assert [c.args[0].split("akcia=")[1] for c in calls] == ["rezervuj", "confirm"] + assert calls[1].kwargs["data"] == {"visitids": "v1"} + m.get_consultation_slots.assert_called_once_with("b1") + + +@pytest.mark.parametrize("status", ["reserved", "temporary", "unknown"]) +def test_not_free_rejected_before_write(booking, status): + m, start, _, slot = booking + slot.status = status + with pytest.raises(ConsultationSlotTaken): + m.book_consultation_slot("b1", start) + m.edupage.session.post.assert_not_called() + + +def test_closed_window(booking): + m, start, block, _ = booking + block.booking_closes_at = datetime.now(timezone.utc) - timedelta(minutes=1) + with pytest.raises(ConsultationClosed): + m.book_consultation_slot("b1", start) + m.edupage.session.post.assert_not_called() + + +def test_existing_own_reservation_not_replaced(booking): + m, start, _, slot = booking + slot.own = True + with pytest.raises(ConsultationLimitReached): + m.book_consultation_slot("b1", start) + + +def test_timeout_no_retry_and_no_confirmation(booking): + m, start, _, _ = booking + m.edupage.session.post.side_effect = Timeout() + with pytest.raises(ConsultationUncertain): + m.book_consultation_slot("b1", start) + assert m.edupage.session.post.call_count == 1 + + +def test_confirmation_must_be_verified(booking): + m, start, _, _ = booking + m.get_consultation_slots.return_value = [] + with pytest.raises(ConsultationUncertain): + m.book_consultation_slot("b1", start) + assert m.edupage.session.post.call_count == 2 + + +def test_expired_session(booking): + m, start, _, _ = booking + m.edupage.session.post.side_effect = None + m.edupage.session.post.return_value = Mock( + status_code=200, url="https://school.edupage.org/login/", text="" + ) + with pytest.raises(ExpiredSessionException): + m.book_consultation_slot("b1", start) + + +def test_cancel_requires_own_id(booking): + m, _, _, _ = booking + with pytest.raises(ConsultationNotOwned): + m.cancel_consultation("foreign") + m.edupage.session.post.assert_not_called() + + +def test_cancel_dry_run_and_actual_verified(booking): + m, start, _, slot = booking + m.get_consultations.return_value.reservations = [ + ConsultationReservation("b1", start, "v1") + ] + slot.own, slot.status, slot.visit_id = True, "reserved", "v1" + plan = m.cancel_consultation("v1", dry_run=True) + assert plan["cancel"] == {"konzultaciaid": "b1"} + m.edupage.session.post.assert_not_called() + m.get_consultation_slots.return_value = [] + m.cancel_consultation("v1") + assert "akcia=zrus" in m.edupage.session.post.call_args.args[0] + assert m.edupage.session.post.call_count == 1 + + +def test_cancel_refuses_ambiguous_block(booking): + m, start, _, slot = booking + m.get_consultations.return_value.reservations = [ + ConsultationReservation("b1", start, "v1") + ] + slot.own, slot.visit_id = True, "v1" + m._detail.return_value[1].append( + ConsultationSlot("b1", start + timedelta(minutes=30), "reserved", True, "v2") + ) + with pytest.raises(ConsultationNotOwned): + m.cancel_consultation("v1") + m.edupage.session.post.assert_not_called() + + +def test_aware_time_and_note_validation(booking): + m, start, _, _ = booking + with pytest.raises(ValueError): + m.book_consultation_slot("b1", start.replace(tzinfo=None)) + with pytest.raises(ConsultationError) as error: + m.book_consultation_slot("b1", start, note="Please call") + assert error.value.code == "note_unsupported" + + +def test_extract_dynamic_form_does_not_execute_script(booking): + m, _, _, _ = booking + # Exercise real parser, including escaped URL delimiters. + del m._detail + m._request = Mock( + return_value=Mock( + text="""""" + ) + ) + path, slots = m._detail("b1") + assert "jwid=hello" in path + assert slots == [] + + +def test_nonexistent_time_is_not_reported_as_taken(booking): + from edupage_api.exceptions import ConsultationSlotNotFound + + m, start, _, _ = booking + with pytest.raises(ConsultationSlotNotFound) as error: + m.book_consultation_slot("b1", start + timedelta(minutes=7), dry_run=True) + assert error.value.code == "slot_not_found" + m.edupage.session.post.assert_not_called() diff --git a/tests/test_grades.py b/tests/test_grades.py index 0022269..4d291bf 100644 --- a/tests/test_grades.py +++ b/tests/test_grades.py @@ -19,10 +19,11 @@ def get_grade(self, details, data="1"): "edupage": {"10": {"p_meno": "Test", "PredmetID": "-5", **details}} }, } - with patch.object( - Grades, "_Grades__get_grade_data", return_value=grade_data - ), patch( - "edupage_api.grades.DbiHelper.fetch_subject_name", return_value="Math" + with ( + patch.object(Grades, "_Grades__get_grade_data", return_value=grade_data), + patch( + "edupage_api.grades.DbiHelper.fetch_subject_name", return_value="Math" + ), ): grades = Grades(self.api).get_grades(None, None) @@ -61,3 +62,155 @@ def test_unknown_grade_type_without_importance(self): if __name__ == "__main__": unittest.main() + + +class MissingGradeMetadataTests(unittest.TestCase): + def setUp(self): + self.api = Edupage() + self.api.is_logged_in = True + + def test_numeric_event_ids_match_json_object_keys(self): + payload = { + "vsetkyZnamky": [ + {"udalostid": 10, "datum": "2026-10-21 10:00:00", "data": "1"} + ], + "vsetkyUdalosti": { + "edupage": { + "10": { + "p_meno": "Test", + "PredmetID": "1", + "p_typ_udalosti": "1", + "p_vaha": "20", + } + } + }, + } + with ( + patch.object(Grades, "_Grades__get_grade_data", return_value=payload), + patch( + "edupage_api.grades.DbiHelper.fetch_subject_name", + return_value="Subject", + ), + ): + result = Grades(self.api).get_grades(None, None) + self.assertEqual(result[0].grade_n, 1) + + def test_missing_event_does_not_silently_change_average(self): + from edupage_api.exceptions import FailedToParseGradeDataError + + payload = { + "vsetkyZnamky": [{"udalostid": "10"}], + "vsetkyUdalosti": {"edupage": {}}, + } + with patch.object(Grades, "_Grades__get_grade_data", return_value=payload): + with self.assertRaisesRegex( + FailedToParseGradeDataError, r"vsetkyUdalosti.edupage\[10\]" + ): + Grades(self.api).get_grades(None, None) + + def test_empty_grades_do_not_require_event_metadata(self): + with patch.object( + Grades, + "_Grades__get_grade_data", + return_value={"vsetkyZnamky": [], "vsetkyUdalosti": None}, + ): + self.assertEqual(Grades(self.api).get_grades(None, None), []) + + def test_missing_event_collection_has_concrete_error(self): + from edupage_api.exceptions import FailedToParseGradeDataError + + with patch.object( + Grades, + "_Grades__get_grade_data", + return_value={ + "vsetkyZnamky": [{"udalostid": "10"}], + "vsetkyUdalosti": None, + }, + ): + with self.assertRaisesRegex( + FailedToParseGradeDataError, "vsetkyUdalosti object" + ): + Grades(self.api).get_grades(None, None) + + +class GradeSkipLoggingTests(unittest.TestCase): + def test_skip_reasons_and_counts_without_grade_values(self): + api = Edupage() + api.is_logged_in = True + payload = { + "vsetkyZnamky": [ + {"data": "PRIVATE_VALUE"}, + {"udalostid": "11", "datum": "2026-10-19 13:16:00", "data": "PRIVATE_VALUE"}, + {"udalostid": "12", "datum": "2026-10-19 13:16:00", "data": "PRIVATE_VALUE"}, + ], + "vsetkyUdalosti": {"edupage": {"11": {}, "12": {"PredmetID": "vsetky"}}}, + } + with patch.object(Grades, "_Grades__get_grade_data", return_value=payload), self.assertLogs("edupage_api.grades", level="DEBUG") as captured: + self.assertEqual(Grades(api).get_grades(None, None), []) + log = " ".join(captured.output) + for reason in ("keys=['data']", "missing_subject_id", "all_subjects", "received=3 parsed=0 skipped=3"): + self.assertIn(reason, log) + self.assertNotIn("PRIVATE_VALUE", log) + + +class StandaloneGradeTests(unittest.TestCase): + def test_standalone_and_event_grades_preserve_count_and_identity(self): + api = Edupage() + api.is_logged_in = True + payload = { + "vsetkyZnamky": [ + {"udalostid": 0, "znamkaid": "42", "predmetid": -5, "datum": "2026-10-19 13:16:00", "data": "A", "p_vaha": "40"}, + {"udalostid": "0", "znamkaid": "43", "predmetid": -5, "datum": "2026-10-20 13:16:00", "data": "(NESPLNĚNO)"}, + {"udalostid": "10", "datum": "2026-10-19 13:16:00", "data": "B"}, + ], + "vsetkyUdalosti": {"edupage": {"10": {"PredmetID": "-5", "p_typ_udalosti": "4", "p_vaha": "20"}}}, + } + with patch.object(Grades, "_Grades__get_grade_data", return_value=payload), patch("edupage_api.grades.DbiHelper.fetch_subject_name", return_value="Inf"): + result = Grades(api).get_grades(None, None) + self.assertEqual([grade.grade_n for grade in result], ["A", "(NESPLNĚNO)", "B"]) + self.assertEqual(result[0].grade_id, "42") + self.assertIsNone(result[0].event_id) + self.assertEqual(result[0].importance, 2) + self.assertEqual(api.grade_read_report, {"received": 3, "parsed": 3, "skipped": 0, "skip_reasons": {}}) + + +class ObservedStandaloneSchemaTests(unittest.TestCase): + def test_thirteen_rows_keep_four_zero_event_grades_for_current_and_term(self): + """Anonymized observed schema: nine event marks + four standalone marks.""" + from edupage_api.grades import Term + api = Edupage() + api.is_logged_in = True + rows = [] + for index in range(13): + rows.append({ + "provider": "edupage", "znamkaid": 100 + index, + "studentid": -1, "predmetid": -5, + "udalostid": 0 if index in (0, 2, 3, 11) else 200 + index, + "data": "A", "datum": "2026-01-01 08:00:00", "ucitelid": -2, + "podpisane": None, "podpisane_rodic": None, + }) + payload = {"vsetkyZnamky": rows, "vsetkyUdalosti": {"edupage": { + str(row["udalostid"]): {"PredmetID": -5, "p_meno": "Paper", "p_typ_udalosti": "4", "p_vaha": 20} + for row in rows if row["udalostid"] + }}} + for term in (None, Term.FIRST): + with self.subTest(term=term), patch.object(Grades, "_Grades__get_grade_data", return_value=payload), patch.object(Grades, "_Grades__get_grade_data_for_term", return_value=payload), patch("edupage_api.grades.DbiHelper.fetch_subject_name", return_value="Subject"), patch("edupage_api.grades.DbiHelper.fetch_teacher_data", return_value=None): + result = Grades(api).get_grades(term, 2026 if term else None) + self.assertEqual(len(result), 13) + self.assertEqual(sum(mark.event_id is None for mark in result), 4) + self.assertEqual(result[11].grade_id, "111") + self.assertEqual(result[11].importance, 1) + self.assertEqual(api.grade_read_report["skipped"], 0) + + +class GradeReportAtomicityTests(unittest.TestCase): + def test_failed_read_does_not_replace_last_successful_report(self): + from edupage_api.exceptions import FailedToParseGradeDataError + api = Edupage() + api.is_logged_in = True + previous = {"received": 7, "parsed": 7, "skipped": 0, "skip_reasons": {}} + api.grade_read_report = previous + payload = {"vsetkyZnamky": [{"udalostid": 10}], "vsetkyUdalosti": {"edupage": {}}} + with patch.object(Grades, "_Grades__get_grade_data", return_value=payload), self.assertRaises(FailedToParseGradeDataError): + Grades(api).get_grades(None, None) + self.assertIs(api.grade_read_report, previous) diff --git a/tests/test_registrations.py b/tests/test_registrations.py new file mode 100644 index 0000000..b3d15d1 --- /dev/null +++ b/tests/test_registrations.py @@ -0,0 +1,259 @@ +"""Synthetic fixtures based on observed response shapes; no account data.""" + +import json +import unittest +from unittest.mock import Mock + +from edupage_api import Edupage +from edupage_api.exceptions import ( + ExpiredSessionException, + MissingDataException, + NotLoggedInException, +) +from edupage_api.registrations import Registrations + + +def html(widget, data): + return ( + '" + ) + + +def response(text="", data=None, status=200, url="https://school.edupage.org/signin/"): + return Mock(text=text, status_code=status, url=url, json=Mock(return_value=data)) + + +class RegistrationTests(unittest.TestCase): + def setUp(self): + self.api = Edupage() + self.api.is_logged_in = True + self.api.subdomain = "school" + self.api.session = Mock() + self.module = Registrations(self.api) + + def set_responses(self, *responses): + self.api.session.post.side_effect = responses + + def bootstrap(self): + return response( + html("signinHome", {"formurl": "/signin/?jwid=dynamic&cmd=SigninHome"}) + ) + + def detail(self, **values): + data = { + "actionData": { + "actionid": "a/1", + "name": "Survey ); { }", + "time_from": "2026-10-01 09:00:00", + "time_to": "2026-11-01 18:00:00", + "active": True, + "targets": [ + { + "targetid": "q1", + "name": "Question", + "data": { + "type": "card", + "content": {"type": "multiple", "text": "Pick"}, + }, + } + ], + }, + "votable": True, + "selection": {}, + } + data.update(values) + return response(html("signinactionplayer", data)) + + def test_empty_assigned_list(self): + self.set_responses(self.bootstrap(), response(data={"actions": []})) + self.assertEqual(self.api.get_registrations(), []) + args, kwargs = self.api.session.post.call_args + self.assertIn("akcia=getData", args[0]) + self.assertEqual(kwargs["data"], {"pohlad": "openableActions"}) + + def test_list_deduplicates_and_loads_closed_detail(self): + closed = {"actionid": "1", "name": "Closed", "active": False, "targets": []} + self.set_responses( + self.bootstrap(), + response(data={"actions": [{"actionid": "1"}, {"actionid": "1"}]}), + response( + html("signinactionplayer", {"actionData": closed, "votable": False}) + ), + ) + result = self.module.get_registrations() + self.assertEqual(len(result), 1) + self.assertFalse(result[0].active) + self.assertEqual(self.api.session.post.call_count, 3) + + def test_survey_content_and_json_answers(self): + self.set_responses( + self.detail( + selection={"q1": {"stav": "answered", "answers": '{"choice": [2]}'}} + ) + ) + result = self.api.get_registration("a/1") + self.assertEqual(result.targets[0].question["type"], "multiple") + self.assertEqual(result.selection[0].answers, {"choice": [2]}) + self.assertEqual(result.opens_at.utcoffset().total_seconds(), 7200) + self.assertEqual(result.closes_at.utcoffset().total_seconds(), 3600) + self.assertIn("actionid=a%2F1", self.api.session.post.call_args.args[0]) + + def test_confirmed_parent_and_temporary_selection(self): + self.set_responses( + self.detail( + parentConfirmed=True, + selection={ + "1": { + "stav": "ok", + "parent_confirmed": "2026-10-01 10:00:00", + "answers": None, + }, + "2": {"stav": "reserved", "reserved_till": "2026-10-01 10:05:00"}, + }, + ) + ) + result = self.module.get_registration("1") + self.assertTrue(result.parent_confirmed) + self.assertEqual(result.selection[1].status, "reserved") + self.assertIsNotNone(result.selection[1].reserved_until) + + def test_history(self): + self.set_responses( + self.bootstrap(), + response( + data={ + "history": [ + { + "actionid": "1", + "action_name": "Action", + "timestamp": "2026-10-01 09:00:00", + } + ] + } + ), + ) + self.assertEqual(self.api.get_registration_history()[0].name, "Action") + self.assertEqual( + self.api.session.post.call_args.kwargs["data"], {"pohlad": "history"} + ) + + def test_missing_list_is_failure_not_empty(self): + self.set_responses(self.bootstrap(), response(data={})) + with self.assertRaises(MissingDataException): + self.module.get_registrations() + + def test_login_expiration(self): + self.set_responses( + response('', url="https://school.edupage.org/login/") + ) + with self.assertRaises(ExpiredSessionException): + self.module.get_registrations() + + def test_not_logged_in(self): + self.api.is_logged_in = False + with self.assertRaises(NotLoggedInException): + self.api.get_consultations() + self.api.session.post.assert_not_called() + + def test_unexpected_html(self): + self.set_responses(response("maintenance")) + with self.assertRaises(MissingDataException): + self.module.get_consultations() + + def test_external_bootstrap_url_rejected(self): + self.set_responses( + response( + html("signinHome", {"formurl": "https://other.test/signin/?jwid=x"}) + ) + ) + with self.assertRaises(MissingDataException): + self.module.get_registrations() + self.assertEqual(self.api.session.post.call_count, 1) + + def test_mutations_and_external_hosts_rejected(self): + for path in ( + "/signin/?cmd=SigninActionPlayer&akcia=confirm", + "/calendar/?cmd=KonzultacneHodinyVisit&akcia=rezervuj", + "/calendar/?cmd=KonzultacneHodinyVisit&akcia=zrus", + "https://other.test/signin/?cmd=SigninHome", + ): + with self.assertRaises(ValueError): + self.module._request(path) + self.api.session.post.assert_not_called() + + def test_consultation_overview_and_dedup_own_reservations(self): + booking = { + "konzultaciaid": "2", + "cas_odkedy": "2026-11-10 13:00:00", + "visitid": "v1", + } + row = { + "konzultaciaid": "2", + "ucitelid": "-1", + "p_meno": "Example", + "p_priezvisko": "Teacher", + "cas_od": "2026-11-10 13:00:00", + "cas_do": "2026-11-10 15:00:00", + "timeframe": "30", + } + self.set_responses( + response( + html( + "konzultacneHodinyScheduler2", + { + "hodinyVsetci": { + "2026-11-10": [row, {**row, "konzultaciaid": "3"}] + }, + "rezervacie": [booking], + "rezervacieMoje": [booking], + }, + ) + ) + ) + result = self.api.get_consultations() + self.assertEqual(len(result.consultations), 2) + self.assertEqual(len(result.reservations), 1) + self.assertIsNone(result.consultations[0].booking_closes_at) + + def test_empty_consultations(self): + self.set_responses( + response( + html( + "konzultacneHodinyScheduler2", + {"hodinyVsetci": {}, "rezervacie": []}, + ) + ) + ) + result = self.module.get_consultations() + self.assertEqual(result.consultations, []) + self.assertEqual(result.reservations, []) + + def test_slots_and_unknown_state(self): + # Free state observed live; occupied variants are synthetic contract tests. + states = [ + "free", + "confirmed", + "confirmed moje", + "unconfirmed moje", + "mystery closed", + ] + slots = "".join( + f'
' + for i, state in enumerate(states) + ) + self.set_responses(response("" + slots)) + result = self.api.get_consultation_slots("3") + self.assertEqual( + [s.status for s in result], + ["free", "reserved", "reserved", "temporary", "unknown"], + ) + self.assertTrue(result[2].own) + self.assertTrue(result[4].closed) + self.assertEqual(result[0].starts_at.isoformat(), "2026-11-10T12:00:00+00:00") + self.assertEqual(self.api.session.post.call_count, 1) + self.assertNotIn("akcia=", self.api.session.post.call_args.args[0]) From 6b80ba39504444397e6569f43874a81b1af6cbdc Mon Sep 17 00:00:00 2001 From: Michal Medvecky <12493312+michalmedvecky@users.noreply.github.com> Date: Thu, 22 Oct 2026 13:22:56 +0200 Subject: [PATCH 2/2] Use unittest for consultation booking tests to match project CI --- tests/test_consultation_booking.py | 285 +++++++++++++++-------------- 1 file changed, 144 insertions(+), 141 deletions(-) diff --git a/tests/test_consultation_booking.py b/tests/test_consultation_booking.py index 0496ab7..0743d53 100644 --- a/tests/test_consultation_booking.py +++ b/tests/test_consultation_booking.py @@ -1,9 +1,9 @@ """Synthetic write protocol tests; no real school requests.""" +import unittest from datetime import datetime, timedelta, timezone from unittest.mock import Mock -import pytest from edupage_api import Edupage from edupage_api.consultation_booking import ConsultationBooking from edupage_api.exceptions import ( @@ -24,10 +24,9 @@ from requests import Timeout -@pytest.fixture -def booking(): +def make_booking(): api = Edupage() - api.subdomain, api.is_logged_in = "school", True + api.subdomain, api.is_logged_in = ("school", True) api.session = Mock() module = ConsultationBooking(api) future = datetime.now(timezone.utc).replace(microsecond=0) + timedelta(days=3) @@ -64,142 +63,146 @@ def booking(): json=Mock(return_value={"status": "ok"}), ), ] - return module, future, block, slot - - -def test_dry_run_performs_no_write(booking): - m, start, _, _ = booking - plan = m.book_consultation_slot("b1", start, dry_run=True) - assert plan["reserve"]["cas"] == str(int(start.timestamp())) - m.edupage.session.post.assert_not_called() - - -def test_book_holds_confirms_and_verifies(booking): - m, start, _, _ = booking - result = m.book_consultation_slot("b1", start) - assert result.visit_id == "v1" - calls = m.edupage.session.post.call_args_list - assert [c.args[0].split("akcia=")[1] for c in calls] == ["rezervuj", "confirm"] - assert calls[1].kwargs["data"] == {"visitids": "v1"} - m.get_consultation_slots.assert_called_once_with("b1") - - -@pytest.mark.parametrize("status", ["reserved", "temporary", "unknown"]) -def test_not_free_rejected_before_write(booking, status): - m, start, _, slot = booking - slot.status = status - with pytest.raises(ConsultationSlotTaken): - m.book_consultation_slot("b1", start) - m.edupage.session.post.assert_not_called() - - -def test_closed_window(booking): - m, start, block, _ = booking - block.booking_closes_at = datetime.now(timezone.utc) - timedelta(minutes=1) - with pytest.raises(ConsultationClosed): - m.book_consultation_slot("b1", start) - m.edupage.session.post.assert_not_called() - - -def test_existing_own_reservation_not_replaced(booking): - m, start, _, slot = booking - slot.own = True - with pytest.raises(ConsultationLimitReached): - m.book_consultation_slot("b1", start) - - -def test_timeout_no_retry_and_no_confirmation(booking): - m, start, _, _ = booking - m.edupage.session.post.side_effect = Timeout() - with pytest.raises(ConsultationUncertain): - m.book_consultation_slot("b1", start) - assert m.edupage.session.post.call_count == 1 - - -def test_confirmation_must_be_verified(booking): - m, start, _, _ = booking - m.get_consultation_slots.return_value = [] - with pytest.raises(ConsultationUncertain): - m.book_consultation_slot("b1", start) - assert m.edupage.session.post.call_count == 2 - - -def test_expired_session(booking): - m, start, _, _ = booking - m.edupage.session.post.side_effect = None - m.edupage.session.post.return_value = Mock( - status_code=200, url="https://school.edupage.org/login/", text="" - ) - with pytest.raises(ExpiredSessionException): - m.book_consultation_slot("b1", start) - - -def test_cancel_requires_own_id(booking): - m, _, _, _ = booking - with pytest.raises(ConsultationNotOwned): - m.cancel_consultation("foreign") - m.edupage.session.post.assert_not_called() - - -def test_cancel_dry_run_and_actual_verified(booking): - m, start, _, slot = booking - m.get_consultations.return_value.reservations = [ - ConsultationReservation("b1", start, "v1") - ] - slot.own, slot.status, slot.visit_id = True, "reserved", "v1" - plan = m.cancel_consultation("v1", dry_run=True) - assert plan["cancel"] == {"konzultaciaid": "b1"} - m.edupage.session.post.assert_not_called() - m.get_consultation_slots.return_value = [] - m.cancel_consultation("v1") - assert "akcia=zrus" in m.edupage.session.post.call_args.args[0] - assert m.edupage.session.post.call_count == 1 - - -def test_cancel_refuses_ambiguous_block(booking): - m, start, _, slot = booking - m.get_consultations.return_value.reservations = [ - ConsultationReservation("b1", start, "v1") - ] - slot.own, slot.visit_id = True, "v1" - m._detail.return_value[1].append( - ConsultationSlot("b1", start + timedelta(minutes=30), "reserved", True, "v2") - ) - with pytest.raises(ConsultationNotOwned): + return (module, future, block, slot) + + +class ConsultationBookingTests(unittest.TestCase): + def test_dry_run_performs_no_write(self): + booking = make_booking() + m, start, _, _ = booking + plan = m.book_consultation_slot("b1", start, dry_run=True) + self.assertEqual(plan["reserve"]["cas"], str(int(start.timestamp()))) + m.edupage.session.post.assert_not_called() + + def test_book_holds_confirms_and_verifies(self): + booking = make_booking() + m, start, _, _ = booking + result = m.book_consultation_slot("b1", start) + self.assertEqual(result.visit_id, "v1") + calls = m.edupage.session.post.call_args_list + self.assertEqual( + [c.args[0].split("akcia=")[1] for c in calls], ["rezervuj", "confirm"] + ) + self.assertEqual(calls[1].kwargs["data"], {"visitids": "v1"}) + m.get_consultation_slots.assert_called_once_with("b1") + + def test_not_free_rejected_before_write(self): + for status in ["reserved", "temporary", "unknown"]: + with self.subTest(status=status): + booking = make_booking() + m, start, _, slot = booking + slot.status = status + with self.assertRaises(ConsultationSlotTaken): + m.book_consultation_slot("b1", start) + m.edupage.session.post.assert_not_called() + + def test_closed_window(self): + booking = make_booking() + m, start, block, _ = booking + block.booking_closes_at = datetime.now(timezone.utc) - timedelta(minutes=1) + with self.assertRaises(ConsultationClosed): + m.book_consultation_slot("b1", start) + m.edupage.session.post.assert_not_called() + + def test_existing_own_reservation_not_replaced(self): + booking = make_booking() + m, start, _, slot = booking + slot.own = True + with self.assertRaises(ConsultationLimitReached): + m.book_consultation_slot("b1", start) + + def test_timeout_no_retry_and_no_confirmation(self): + booking = make_booking() + m, start, _, _ = booking + m.edupage.session.post.side_effect = Timeout() + with self.assertRaises(ConsultationUncertain): + m.book_consultation_slot("b1", start) + self.assertEqual(m.edupage.session.post.call_count, 1) + + def test_confirmation_must_be_verified(self): + booking = make_booking() + m, start, _, _ = booking + m.get_consultation_slots.return_value = [] + with self.assertRaises(ConsultationUncertain): + m.book_consultation_slot("b1", start) + self.assertEqual(m.edupage.session.post.call_count, 2) + + def test_expired_session(self): + booking = make_booking() + m, start, _, _ = booking + m.edupage.session.post.side_effect = None + m.edupage.session.post.return_value = Mock( + status_code=200, url="https://school.edupage.org/login/", text="" + ) + with self.assertRaises(ExpiredSessionException): + m.book_consultation_slot("b1", start) + + def test_cancel_requires_own_id(self): + booking = make_booking() + m, _, _, _ = booking + with self.assertRaises(ConsultationNotOwned): + m.cancel_consultation("foreign") + m.edupage.session.post.assert_not_called() + + def test_cancel_dry_run_and_actual_verified(self): + booking = make_booking() + m, start, _, slot = booking + m.get_consultations.return_value.reservations = [ + ConsultationReservation("b1", start, "v1") + ] + slot.own, slot.status, slot.visit_id = (True, "reserved", "v1") + plan = m.cancel_consultation("v1", dry_run=True) + self.assertEqual(plan["cancel"], {"konzultaciaid": "b1"}) + m.edupage.session.post.assert_not_called() + m.get_consultation_slots.return_value = [] m.cancel_consultation("v1") - m.edupage.session.post.assert_not_called() - - -def test_aware_time_and_note_validation(booking): - m, start, _, _ = booking - with pytest.raises(ValueError): - m.book_consultation_slot("b1", start.replace(tzinfo=None)) - with pytest.raises(ConsultationError) as error: - m.book_consultation_slot("b1", start, note="Please call") - assert error.value.code == "note_unsupported" - - -def test_extract_dynamic_form_does_not_execute_script(booking): - m, _, _, _ = booking - # Exercise real parser, including escaped URL delimiters. - del m._detail - m._request = Mock( - return_value=Mock( - text="""""" + self.assertIn("akcia=zrus", m.edupage.session.post.call_args.args[0]) + self.assertEqual(m.edupage.session.post.call_count, 1) + + def test_cancel_refuses_ambiguous_block(self): + booking = make_booking() + m, start, _, slot = booking + m.get_consultations.return_value.reservations = [ + ConsultationReservation("b1", start, "v1") + ] + slot.own, slot.visit_id = (True, "v1") + m._detail.return_value[1].append( + ConsultationSlot( + "b1", start + timedelta(minutes=30), "reserved", True, "v2" + ) ) - ) - path, slots = m._detail("b1") - assert "jwid=hello" in path - assert slots == [] - - -def test_nonexistent_time_is_not_reported_as_taken(booking): - from edupage_api.exceptions import ConsultationSlotNotFound - - m, start, _, _ = booking - with pytest.raises(ConsultationSlotNotFound) as error: - m.book_consultation_slot("b1", start + timedelta(minutes=7), dry_run=True) - assert error.value.code == "slot_not_found" - m.edupage.session.post.assert_not_called() + with self.assertRaises(ConsultationNotOwned): + m.cancel_consultation("v1") + m.edupage.session.post.assert_not_called() + + def test_aware_time_and_note_validation(self): + booking = make_booking() + m, start, _, _ = booking + with self.assertRaises(ValueError): + m.book_consultation_slot("b1", start.replace(tzinfo=None)) + with self.assertRaises(ConsultationError) as error: + m.book_consultation_slot("b1", start, note="Please call") + self.assertEqual(error.exception.code, "note_unsupported") + + def test_extract_dynamic_form_does_not_execute_script(self): + booking = make_booking() + m, _, _, _ = booking + del m._detail + m._request = Mock( + return_value=Mock( + text="" + ) + ) + path, slots = m._detail("b1") + self.assertIn("jwid=hello", path) + self.assertEqual(slots, []) + + def test_nonexistent_time_is_not_reported_as_taken(self): + booking = make_booking() + from edupage_api.exceptions import ConsultationSlotNotFound + + m, start, _, _ = booking + with self.assertRaises(ConsultationSlotNotFound) as error: + m.book_consultation_slot("b1", start + timedelta(minutes=7), dry_run=True) + self.assertEqual(error.exception.code, "slot_not_found") + m.edupage.session.post.assert_not_called()