from __future__ import annotations import unittest from datetime import datetime, timedelta, timezone from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from govoplan_access.backend.db import models as access_models # noqa: F401 from govoplan_calendar.backend.capabilities import ( SqlCalendarExternalProfileProvider, SqlCalendarInvitationProvider, SqlCalendarSchedulingProvider, ) from govoplan_calendar.backend.db.models import CalendarEvent from govoplan_calendar.backend.schemas import CalendarCollectionCreateRequest from govoplan_calendar.backend.service import create_calendar from govoplan_core.core.calendar import ( CalendarCapabilityError, CalendarEventRequest, CalendarExternalProfileRequest, CalendarInvitationAttendeeRequest, CalendarInvitationRequest, ) from govoplan_core.db.base import Base from govoplan_core.tenancy.scope import create_scope_tables from govoplan_tenancy.backend.db.models import Tenant class CalendarSchedulingCapabilityTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite:///:memory:") create_scope_tables(self.engine) Base.metadata.create_all(bind=self.engine) self.Session = sessionmaker(bind=self.engine) self.session = self.Session() self.session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) self.calendar = create_calendar( self.session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Scheduling"), ) self.provider = SqlCalendarSchedulingProvider() def tearDown(self) -> None: self.session.close() Base.metadata.drop_all(bind=self.engine) self.engine.dispose() def test_event_request_validation_is_exposed_as_capability_error(self) -> None: with self.assertRaises(CalendarCapabilityError): self.provider.create_event( object(), tenant_id="tenant-1", user_id="user-1", request=CalendarEventRequest( summary="x" * 501, start_at=datetime(2026, 7, 20, 9, tzinfo=timezone.utc), ), ) def test_hold_promotion_and_release_are_idempotent(self) -> None: start = datetime(2026, 7, 20, 9, tzinfo=timezone.utc) hold = self.provider.create_event( self.session, tenant_id="tenant-1", user_id="user-1", request=CalendarEventRequest( calendar_id=self.calendar.id, summary="Tentative hold", status="TENTATIVE", start_at=start, end_at=start + timedelta(hours=1), metadata={"scheduling_request_id": "request-1"}, ), ) promoted = self.provider.promote_event( self.session, tenant_id="tenant-1", user_id="user-1", event_id=hold.id, request=CalendarEventRequest( calendar_id=self.calendar.id, summary="Confirmed meeting", status="CONFIRMED", start_at=start, end_at=start + timedelta(hours=1), metadata={"scheduling_request_id": "request-1"}, ), ) released = self.provider.release_event( self.session, tenant_id="tenant-1", user_id="user-1", event_id=promoted.id, ) replayed = self.provider.release_event( self.session, tenant_id="tenant-1", user_id="user-1", event_id=promoted.id, ) absent = self.provider.release_event( self.session, tenant_id="tenant-1", user_id="user-1", event_id="missing-event", ) self.assertEqual(promoted.id, hold.id) self.assertEqual("CONFIRMED", self.session.get(CalendarEvent, hold.id).status) self.assertFalse(released.already_released) self.assertTrue(replayed.already_released) self.assertEqual("not_found", absent.external_state) class CalendarExternalProfileCapabilityTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite:///:memory:") create_scope_tables(self.engine) Base.metadata.create_all(bind=self.engine) self.Session = sessionmaker(bind=self.engine) self.session = self.Session() self.session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) self.calendar = create_calendar( self.session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Open-Xchange"), ) self.provider = SqlCalendarExternalProfileProvider() def tearDown(self) -> None: self.session.close() Base.metadata.drop_all(bind=self.engine) self.engine.dispose() def test_open_xchange_profile_uses_caldav_and_retains_mapping_refs(self) -> None: configured = self.provider.configure_profile( self.session, tenant_id="tenant-1", user_id=None, request=CalendarExternalProfileRequest( profile_kind="open-xchange", calendar_id=self.calendar.id, endpoint_url="https://groupware.example.test/caldav/team/", connector_profile_ref="connector:ox-main", identity_mapping_ref="idm:ox-main", resource_calendar_ref="ox-resource:room-42", ), ) source = self.calendar.metadata_["sync_source_id"] self.session.refresh(self.calendar) self.assertEqual(configured.source_id, source) self.assertEqual("caldav", configured.transport_kind) self.assertEqual("resource", self.calendar.owner_type) self.assertEqual("ox-resource:room-42", self.calendar.owner_id) self.assertEqual("open_xchange", self.calendar.metadata_["integration_profile"]) def test_unsupported_profile_is_rejected(self) -> None: with self.assertRaisesRegex(CalendarCapabilityError, "Unsupported"): self.provider.configure_profile( self.session, tenant_id="tenant-1", user_id=None, request=CalendarExternalProfileRequest( profile_kind="unknown", calendar_id=self.calendar.id, endpoint_url="https://groupware.example.test/caldav/team/", ), ) class CalendarInvitationCapabilityTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite:///:memory:") create_scope_tables(self.engine) Base.metadata.create_all(bind=self.engine) self.Session = sessionmaker(bind=self.engine) self.session = self.Session() self.session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) self.calendar = create_calendar( self.session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Invitations"), ) self.provider = SqlCalendarInvitationProvider() def tearDown(self) -> None: self.session.close() Base.metadata.drop_all(bind=self.engine) self.engine.dispose() def request(self, *, summary: str = "Planning") -> CalendarInvitationRequest: return CalendarInvitationRequest( correlation_id="campaign-1:recipient-1", source_module="campaigns", source_resource_type="campaign_recipient", source_resource_id="recipient-1", calendar_id=self.calendar.id, summary=summary, start_at=datetime(2026, 8, 5, 9, tzinfo=timezone.utc), end_at=datetime(2026, 8, 5, 10, tzinfo=timezone.utc), organizer={ "value": "mailto:organizer@example.test", "params": {"CN": ["Organizer"]}, }, attendees=( CalendarInvitationAttendeeRequest( address="ada@example.test", name="Ada", ), ), ) def test_upsert_and_response_reconciliation_preserve_correlation(self) -> None: created = self.provider.upsert_invitation( self.session, tenant_id="tenant-1", user_id=None, request=self.request(), ) response = self.provider.record_response( self.session, tenant_id="tenant-1", correlation_id=created.correlation_id, attendee_address="ADA@example.test", participation_status="accepted", evidence={"transport": "ical-reply"}, ) updated = self.provider.upsert_invitation( self.session, tenant_id="tenant-1", user_id=None, request=self.request(summary="Updated planning"), ) loaded = self.provider.get_invitation( self.session, tenant_id="tenant-1", correlation_id=created.correlation_id, ) self.assertEqual(created.event_id, updated.event_id) self.assertEqual("ACCEPTED", response.attendees[0]["params"]["PARTSTAT"][0]) self.assertEqual("ACCEPTED", updated.attendees[0]["params"]["PARTSTAT"][0]) self.assertEqual(updated, loaded) self.assertFalse(updated.recurrence_supported) self.assertTrue(updated.degraded_reasons) def test_invitation_render_batch_lookup_and_summary(self) -> None: request = self.request() payload = self.provider.render_invitation(request) created = self.provider.upsert_invitation( self.session, tenant_id="tenant-1", user_id=None, request=request, ) self.assertIn("METHOD:REQUEST", payload) self.assertIn(f"UID:{created.uid}", payload) self.assertEqual( created, self.provider.get_invitations( self.session, tenant_id="tenant-1", correlation_ids=(created.correlation_id,), )[created.correlation_id], ) summary = self.provider.summarize_invitations( self.session, tenant_id="tenant-1", source_module="campaigns", source_resource_type="campaign_recipient", source_resource_id="recipient-1", ) self.assertEqual(1, summary["invitation_count"]) self.assertEqual(1, summary["pending_response_count"]) calendars = self.provider.list_calendars( self.session, tenant_id="tenant-1", ) self.assertEqual((self.calendar.id,), tuple(item.id for item in calendars)) self.assertTrue(calendars[0].writable) def test_icalendar_reply_updates_attendee_status(self) -> None: created = self.provider.upsert_invitation( self.session, tenant_id="tenant-1", user_id=None, request=self.request(), ) reply = "\r\n".join( ( "BEGIN:VCALENDAR", "VERSION:2.0", "METHOD:REPLY", "BEGIN:VEVENT", f"UID:{created.uid}", "DTSTART:20260805T090000Z", "ATTENDEE;CN=Ada;PARTSTAT=TENTATIVE:mailto:ada@example.test", "END:VEVENT", "END:VCALENDAR", "", ) ) recorded = self.provider.record_icalendar_reply( self.session, tenant_id="tenant-1", icalendar=reply, evidence={"profile_id": "mail-profile-1", "uid": "42"}, ) sequence_after_first_reply = self.session.get( CalendarEvent, created.event_id, ).sequence replayed = self.provider.record_icalendar_reply( self.session, tenant_id="tenant-1", icalendar=reply, evidence={"profile_id": "mail-profile-1", "uid": "42"}, ) self.assertEqual(1, len(recorded)) self.assertEqual(1, len(replayed)) self.assertEqual( "TENTATIVE", recorded[0].attendees[0]["params"]["PARTSTAT"][0], ) self.assertEqual( sequence_after_first_reply, self.session.get(CalendarEvent, created.event_id).sequence, ) def test_reply_must_match_an_existing_attendee(self) -> None: created = self.provider.upsert_invitation( self.session, tenant_id="tenant-1", user_id=None, request=self.request(), ) with self.assertRaisesRegex(CalendarCapabilityError, "not an attendee"): self.provider.record_response( self.session, tenant_id="tenant-1", uid=created.uid, attendee_address="mallory@example.test", participation_status="DECLINED", ) if __name__ == "__main__": unittest.main()