chore: sync GovOPlaN module split state
This commit is contained in:
171
tests/test_sync_sources.py
Normal file
171
tests/test_sync_sources.py
Normal file
@@ -0,0 +1,171 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import unittest
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import patch
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from govoplan_access.backend.db import models as access_models # noqa: F401 - populate users/accounts tables
|
||||
from govoplan_calendar.backend.db.models import CalendarEvent
|
||||
from govoplan_calendar.backend.schemas import CalendarCollectionCreateRequest, CalendarEventCreateRequest, CalendarSyncSourceCreateRequest
|
||||
from govoplan_calendar.backend.service import CalendarError, create_calendar, create_event, create_sync_source, sync_source
|
||||
from govoplan_core.db.base import Base
|
||||
from govoplan_tenancy.backend.db.models import Tenant
|
||||
|
||||
|
||||
class CalendarSyncSourceTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.engine = create_engine("sqlite:///:memory:")
|
||||
Base.metadata.create_all(bind=self.engine)
|
||||
self.Session = sessionmaker(bind=self.engine)
|
||||
|
||||
def tearDown(self) -> None:
|
||||
Base.metadata.drop_all(bind=self.engine)
|
||||
|
||||
def session_with_calendar(self):
|
||||
session = self.Session()
|
||||
session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
|
||||
calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))
|
||||
session.flush()
|
||||
return session, calendar
|
||||
|
||||
def test_ics_subscription_sync_imports_events_and_blocks_local_mutation(self) -> None:
|
||||
session, calendar = self.session_with_calendar()
|
||||
source = create_sync_source(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
user_id=None,
|
||||
payload=CalendarSyncSourceCreateRequest(
|
||||
source_kind="ics",
|
||||
calendar_id=calendar.id,
|
||||
collection_url="https://calendar.example.test/feed.ics",
|
||||
sync_direction="two_way",
|
||||
),
|
||||
)
|
||||
|
||||
ics = """BEGIN:VCALENDAR
|
||||
VERSION:2.0
|
||||
BEGIN:VEVENT
|
||||
UID:ics-1@example.test
|
||||
DTSTART:20260708T090000Z
|
||||
DTEND:20260708T100000Z
|
||||
SUMMARY:ICS item
|
||||
END:VEVENT
|
||||
END:VCALENDAR
|
||||
"""
|
||||
with patch("govoplan_calendar.backend.service.http_request", return_value=(200, {"etag": '"feed-1"'}, ics)):
|
||||
refreshed, stats = sync_source(session, tenant_id="tenant-1", user_id=None, source_id=source.id)
|
||||
|
||||
event = session.query(CalendarEvent).one()
|
||||
self.assertEqual(refreshed.sync_direction, "inbound")
|
||||
self.assertEqual(stats.created, 1)
|
||||
self.assertEqual(event.source_kind, "ics")
|
||||
self.assertEqual(event.summary, "ICS item")
|
||||
with self.assertRaisesRegex(CalendarError, "inbound-only"):
|
||||
create_event(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
user_id=None,
|
||||
payload=CalendarEventCreateRequest(
|
||||
calendar_id=calendar.id,
|
||||
summary="Local edit",
|
||||
start_at=datetime(2026, 7, 8, 12, tzinfo=timezone.utc),
|
||||
),
|
||||
)
|
||||
|
||||
def test_graph_sync_imports_delta_events(self) -> None:
|
||||
session, calendar = self.session_with_calendar()
|
||||
source = create_sync_source(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
user_id=None,
|
||||
payload=CalendarSyncSourceCreateRequest(
|
||||
source_kind="graph",
|
||||
calendar_id=calendar.id,
|
||||
collection_url="me/calendar",
|
||||
auth_type="bearer",
|
||||
bearer_token="graph-token",
|
||||
),
|
||||
)
|
||||
payload = {
|
||||
"value": [
|
||||
{
|
||||
"id": "graph-event-1",
|
||||
"iCalUId": "graph-uid-1",
|
||||
"subject": "Graph item",
|
||||
"start": {"dateTime": "2026-07-08T09:00:00", "timeZone": "UTC"},
|
||||
"end": {"dateTime": "2026-07-08T10:00:00", "timeZone": "UTC"},
|
||||
"@odata.etag": "graph-etag-1",
|
||||
}
|
||||
],
|
||||
"@odata.deltaLink": "https://graph.microsoft.com/v1.0/me/calendar/events/delta?$deltatoken=next",
|
||||
}
|
||||
|
||||
with patch("govoplan_calendar.backend.service.http_request", return_value=(200, {}, json.dumps(payload))) as request:
|
||||
refreshed, stats = sync_source(session, tenant_id="tenant-1", user_id=None, source_id=source.id)
|
||||
|
||||
event = session.query(CalendarEvent).one()
|
||||
self.assertEqual(stats.created, 1)
|
||||
self.assertEqual(refreshed.sync_token, payload["@odata.deltaLink"])
|
||||
self.assertEqual(event.source_kind, "graph")
|
||||
self.assertEqual(event.source_href, "graph-event-1")
|
||||
self.assertEqual(event.summary, "Graph item")
|
||||
self.assertIn("Bearer graph-token", request.call_args.kwargs["headers"]["Authorization"])
|
||||
|
||||
def test_ews_sync_imports_calendar_view_items(self) -> None:
|
||||
session, calendar = self.session_with_calendar()
|
||||
source = create_sync_source(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
user_id=None,
|
||||
payload=CalendarSyncSourceCreateRequest(
|
||||
source_kind="ews",
|
||||
calendar_id=calendar.id,
|
||||
collection_url="https://exchange.example.test/EWS/Exchange.asmx",
|
||||
auth_type="basic",
|
||||
username="ada",
|
||||
password="secret",
|
||||
),
|
||||
)
|
||||
response_xml = """<?xml version="1.0" encoding="utf-8"?>
|
||||
<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/"
|
||||
xmlns:m="http://schemas.microsoft.com/exchange/services/2006/messages"
|
||||
xmlns:t="http://schemas.microsoft.com/exchange/services/2006/types">
|
||||
<s:Body>
|
||||
<m:FindItemResponse>
|
||||
<m:ResponseMessages>
|
||||
<m:FindItemResponseMessage ResponseClass="Success">
|
||||
<m:RootFolder>
|
||||
<t:Items>
|
||||
<t:CalendarItem>
|
||||
<t:ItemId Id="ews-event-1" ChangeKey="ews-etag-1" />
|
||||
<t:Subject>EWS item</t:Subject>
|
||||
<t:Start>2026-07-08T09:00:00Z</t:Start>
|
||||
<t:End>2026-07-08T10:00:00Z</t:End>
|
||||
<t:IsAllDayEvent>false</t:IsAllDayEvent>
|
||||
<t:UID>ews-uid-1</t:UID>
|
||||
</t:CalendarItem>
|
||||
</t:Items>
|
||||
</m:RootFolder>
|
||||
</m:FindItemResponseMessage>
|
||||
</m:ResponseMessages>
|
||||
</m:FindItemResponse>
|
||||
</s:Body>
|
||||
</s:Envelope>"""
|
||||
|
||||
with patch("govoplan_calendar.backend.service.http_request", return_value=(200, {}, response_xml)) as request:
|
||||
_refreshed, stats = sync_source(session, tenant_id="tenant-1", user_id=None, source_id=source.id)
|
||||
|
||||
event = session.query(CalendarEvent).one()
|
||||
self.assertEqual(stats.created, 1)
|
||||
self.assertEqual(event.source_kind, "ews")
|
||||
self.assertEqual(event.source_href, "ews-event-1")
|
||||
self.assertEqual(event.summary, "EWS item")
|
||||
self.assertTrue(request.call_args.kwargs["body"].startswith("<?xml"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user