135 lines
4.8 KiB
Python
135 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime, 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 (
|
|
SqlCalendarInvitationProvider,
|
|
SqlCalendarSchedulingProvider,
|
|
)
|
|
from govoplan_calendar.backend.schemas import CalendarCollectionCreateRequest
|
|
from govoplan_calendar.backend.service import create_calendar
|
|
from govoplan_core.core.calendar import (
|
|
CalendarCapabilityError,
|
|
CalendarEventRequest,
|
|
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 test_event_request_validation_is_exposed_as_capability_error(self) -> None:
|
|
provider = SqlCalendarSchedulingProvider()
|
|
|
|
with self.assertRaises(CalendarCapabilityError):
|
|
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),
|
|
),
|
|
)
|
|
|
|
|
|
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),
|
|
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_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()
|