from __future__ import annotations from datetime import datetime, timedelta, timezone import pytest from govoplan_scheduling.backend.db.models import ( SchedulingCandidateSlot, SchedulingParticipant, SchedulingRequest, ) from govoplan_scheduling.backend.schemas import ( SchedulingCandidateSlotReconcileInput, SchedulingParticipantReconcileInput, SchedulingRequestUpdateRequest, ) from govoplan_scheduling.backend.service import ( SchedulingError, _plan_scheduling_participant_reconciliation, _plan_scheduling_request_update, _plan_scheduling_slot_reconciliation, scheduling_participant_revision, scheduling_slot_revision, ) NOW = datetime(2026, 7, 29, 9, tzinfo=timezone.utc) def _request() -> SchedulingRequest: return SchedulingRequest( id="request-1", tenant_id="tenant-1", title="Steering group", timezone="Europe/Berlin", status="collecting", poll_id="poll-1", allow_external_participants=True, allow_participant_updates=True, result_visibility="after_close", participant_visibility="aggregates_only", notify_on_answers=True, single_choice=False, max_participants_per_option=None, allow_maybe=True, allow_comments=False, participant_email_required=False, anonymous_password_protection_enabled=False, ) def _slot( request: SchedulingRequest, *, slot_id: str, position: int, start_offset: int, ) -> SchedulingCandidateSlot: slot = SchedulingCandidateSlot( id=slot_id, tenant_id=request.tenant_id, request=request, poll_option_id=f"option-{slot_id}", label=f"Slot {position + 1}", start_at=NOW + timedelta(hours=start_offset), end_at=NOW + timedelta(hours=start_offset + 1), timezone=request.timezone, position=position, freebusy_conflicts=[], metadata_={}, ) return slot def _slot_input( slot: SchedulingCandidateSlot, *, label: str | None = None, ) -> SchedulingCandidateSlotReconcileInput: return SchedulingCandidateSlotReconcileInput( id=slot.id, revision=scheduling_slot_revision(slot), label=label or slot.label, start_at=slot.start_at, end_at=slot.end_at, timezone=slot.timezone, location=slot.location, metadata=slot.metadata_ or {}, ) def _participant( request: SchedulingRequest, *, participant_id: str, respondent_id: str, email: str, required: bool, status: str = "invited", invitation_id: str | None = None, ) -> SchedulingParticipant: return SchedulingParticipant( id=participant_id, tenant_id=request.tenant_id, request=request, respondent_id=respondent_id, display_name=participant_id.title(), email=email, participant_type="internal", required=required, status=status, poll_invitation_id=invitation_id, participation_gateway="scheduling" if invitation_id else None, metadata_={}, ) def _participant_input( participant: SchedulingParticipant, **changes: object, ) -> SchedulingParticipantReconcileInput: values = { "id": participant.id, "revision": scheduling_participant_revision(participant), "respondent_id": participant.respondent_id, "display_name": participant.display_name, "email": participant.email, "participant_type": participant.participant_type, "required": participant.required, "metadata": participant.metadata_ or {}, } values.update(changes) return SchedulingParticipantReconcileInput.model_validate(values) def test_slot_plan_is_inspectable_and_does_not_mutate_models() -> None: request = _request() first = _slot(request, slot_id="slot-1", position=0, start_offset=1) second = _slot(request, slot_id="slot-2", position=1, start_offset=3) supplied = [ _slot_input(first, label="Updated first slot"), SchedulingCandidateSlotReconcileInput( label="New slot", start_at=NOW + timedelta(hours=5), end_at=NOW + timedelta(hours=6), timezone=request.timezone, ), ] plan = _plan_scheduling_slot_reconciliation( request=request, supplied_slots=supplied, option_mutation_available=True, ) assert [update.slot_id for update in plan.updates] == ["slot-1"] assert plan.updates[0].changes.label == "Updated first slot" assert len(plan.additions) == 1 assert plan.removals == ("slot-2",) assert plan.changed is True assert first.label == "Slot 1" assert second.deleted_at is None assert len(request.slots) == 2 def test_exact_slot_replay_produces_a_noop_plan() -> None: request = _request() first = _slot(request, slot_id="slot-1", position=0, start_offset=1) plan = _plan_scheduling_slot_reconciliation( request=request, supplied_slots=[_slot_input(first)], option_mutation_available=True, ) assert plan.changed is False assert plan.updates == () assert plan.additions == () assert plan.removals == () def test_slot_plan_rejects_removal_of_a_tentative_calendar_hold() -> None: request = _request() held = _slot(request, slot_id="slot-held", position=0, start_offset=1) held.tentative_hold_event_id = "event-1" with pytest.raises(SchedulingError, match="tentative calendar hold"): _plan_scheduling_slot_reconciliation( request=request, supplied_slots=[], option_mutation_available=True, ) def test_participant_plan_distinguishes_updates_additions_and_retirements() -> None: request = _request() alice = _participant( request, participant_id="alice", respondent_id="user-alice", email="alice@example.test", required=True, invitation_id="invitation-alice", ) bob = _participant( request, participant_id="bob", respondent_id="user-bob", email="bob@example.test", required=False, status="responded", ) supplied = [ _participant_input(alice, email="alice.new@example.test", required=False), SchedulingParticipantReconcileInput( respondent_id="user-charlie", display_name="Charlie", email="charlie@example.test", participant_type="internal", required=True, ), ] plan = _plan_scheduling_participant_reconciliation( request=request, supplied_participants=supplied, participation_available=True, retirement_available=True, ) assert len(plan.updates) == 1 assert plan.updates[0].participant_id == "alice" assert plan.updates[0].revoke_invitation is True assert set(plan.updates[0].changed_fields) == {"email", "required"} assert plan.creations[0].replaces_participant_id is None assert plan.creations[0].supplied.required is True assert plan.retirements == ("bob",) assert alice.email == "alice@example.test" assert alice.required is True assert bob.status == "responded" def test_request_plan_records_invitation_expiry_work_without_tokens() -> None: request = _request() participant = _participant( request, participant_id="alice", respondent_id="user-alice", email="alice@example.test", required=True, invitation_id="invitation-alice", ) deadline = NOW + timedelta(days=2) plan = _plan_scheduling_request_update( request=request, payload=SchedulingRequestUpdateRequest(deadline_at=deadline), participation_available=True, ) assert plan.deadline_changed is True assert plan.retained_invitation_ids == ((participant.id, "invitation-alice"),) assert "token" not in repr(plan).casefold() assert request.deadline_at is None def test_request_plan_rejects_policy_change_after_link_issuance() -> None: request = _request() _participant( request, participant_id="alice", respondent_id="user-alice", email="alice@example.test", required=True, invitation_id="invitation-alice", ) with pytest.raises(SchedulingError, match="cannot change"): _plan_scheduling_request_update( request=request, payload=SchedulingRequestUpdateRequest(single_choice=True), participation_available=True, )