Implement Open-Xchange calendar profiles
This commit is contained in:
@@ -8,6 +8,7 @@ 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,
|
||||
)
|
||||
@@ -16,6 +17,7 @@ from govoplan_calendar.backend.service import create_calendar
|
||||
from govoplan_core.core.calendar import (
|
||||
CalendarCapabilityError,
|
||||
CalendarEventRequest,
|
||||
CalendarExternalProfileRequest,
|
||||
CalendarInvitationAttendeeRequest,
|
||||
CalendarInvitationRequest,
|
||||
)
|
||||
@@ -40,6 +42,64 @@ class CalendarSchedulingCapabilityTests(unittest.TestCase):
|
||||
)
|
||||
|
||||
|
||||
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:")
|
||||
|
||||
@@ -13,6 +13,7 @@ from govoplan_calendar.backend.manifest import (
|
||||
EWS_PROVIDER_ID,
|
||||
GRAPH_PROVIDER_ID,
|
||||
ICS_PROVIDER_ID,
|
||||
OPEN_XCHANGE_PROVIDER_ID,
|
||||
manifest,
|
||||
)
|
||||
from govoplan_calendar.backend.provider_state import (
|
||||
@@ -20,6 +21,7 @@ from govoplan_calendar.backend.provider_state import (
|
||||
ews_provider_states,
|
||||
graph_provider_states,
|
||||
ics_provider_states,
|
||||
open_xchange_provider_states,
|
||||
)
|
||||
from govoplan_calendar.backend.schemas import (
|
||||
CalendarCalDavSourceCreateRequest,
|
||||
@@ -117,11 +119,11 @@ class CalendarProviderStateTests(unittest.TestCase):
|
||||
),
|
||||
)
|
||||
self.assertEqual(
|
||||
{CALDAV_PROVIDER_ID, ICS_PROVIDER_ID, GRAPH_PROVIDER_ID, EWS_PROVIDER_ID},
|
||||
{CALDAV_PROVIDER_ID, ICS_PROVIDER_ID, GRAPH_PROVIDER_ID, EWS_PROVIDER_ID, OPEN_XCHANGE_PROVIDER_ID},
|
||||
{item.id for item in manifest.external_providers},
|
||||
)
|
||||
self.assertEqual(
|
||||
{CALDAV_PROVIDER_ID, ICS_PROVIDER_ID, GRAPH_PROVIDER_ID, EWS_PROVIDER_ID},
|
||||
{CALDAV_PROVIDER_ID, ICS_PROVIDER_ID, GRAPH_PROVIDER_ID, EWS_PROVIDER_ID, OPEN_XCHANGE_PROVIDER_ID},
|
||||
{
|
||||
item.provider_id
|
||||
for item in manifest.external_provider_state_providers
|
||||
@@ -169,6 +171,30 @@ class CalendarProviderStateTests(unittest.TestCase):
|
||||
self.assertNotIn("feeds.example.test", rendered)
|
||||
self.assertNotIn("exchange.example.test", rendered)
|
||||
|
||||
def test_open_xchange_profile_has_distinct_provider_state(self) -> None:
|
||||
self.source.metadata_ = {
|
||||
"integration_profile": "open_xchange",
|
||||
"connector_profile_ref": "connector:secret-detail",
|
||||
"identity_mapping_ref": "idm:secret-detail",
|
||||
}
|
||||
self.source.last_status = "ok"
|
||||
self.source.last_synced_at = datetime.now(UTC)
|
||||
self.session.flush()
|
||||
|
||||
self.assertEqual(
|
||||
(),
|
||||
caldav_provider_states(
|
||||
ExternalProviderStateContext(session=self.session, tenant_id="tenant-1")
|
||||
),
|
||||
)
|
||||
state = open_xchange_provider_states(
|
||||
ExternalProviderStateContext(session=self.session, tenant_id="tenant-1")
|
||||
)[0]
|
||||
|
||||
self.assertEqual(OPEN_XCHANGE_PROVIDER_ID, state.provider_id)
|
||||
self.assertEqual("healthy", state.health)
|
||||
self.assertNotIn("secret-detail", str(state.to_dict()))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user