358 lines
13 KiB
Python
358 lines
13 KiB
Python
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()
|