276 lines
11 KiB
Python
276 lines
11 KiB
Python
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, soft_delete_unseen_source_events, sync_source
|
|
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 CalendarSyncSourceTests(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)
|
|
|
|
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_ics_not_modified_sync_clears_error_and_reschedules(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",
|
|
),
|
|
)
|
|
source.ctag = '"feed-1"'
|
|
source.last_status = "error"
|
|
source.last_error = "previous failure"
|
|
|
|
with patch("govoplan_calendar.backend.service.http_request", return_value=(304, {}, "")):
|
|
refreshed, stats = sync_source(session, tenant_id="tenant-1", user_id=None, source_id=source.id)
|
|
|
|
self.assertEqual(refreshed.last_status, "ok")
|
|
self.assertIsNone(refreshed.last_error)
|
|
self.assertIsNotNone(refreshed.last_synced_at)
|
|
self.assertIsNotNone(refreshed.next_sync_at)
|
|
self.assertEqual(stats.created, 0)
|
|
self.assertEqual(stats.updated, 0)
|
|
self.assertEqual(stats.deleted, 0)
|
|
|
|
def test_ics_sync_failure_records_error_and_reschedules(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",
|
|
),
|
|
)
|
|
|
|
with patch("govoplan_calendar.backend.service.http_request", side_effect=RuntimeError("network down")):
|
|
with self.assertRaisesRegex(RuntimeError, "network down"):
|
|
sync_source(session, tenant_id="tenant-1", user_id=None, source_id=source.id)
|
|
|
|
self.assertEqual(source.last_status, "error")
|
|
self.assertIn("network down", source.last_error or "")
|
|
self.assertIsNotNone(source.last_attempt_at)
|
|
self.assertIsNotNone(source.next_sync_at)
|
|
|
|
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"))
|
|
|
|
def test_full_sync_cleanup_soft_deletes_unseen_events_in_batches(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/events.ics",
|
|
),
|
|
)
|
|
for index in range(3):
|
|
session.add(
|
|
CalendarEvent(
|
|
tenant_id="tenant-1",
|
|
calendar_id=calendar.id,
|
|
uid=f"event-{index}@example.test",
|
|
recurrence_id=None,
|
|
summary=f"Event {index}",
|
|
status="CONFIRMED",
|
|
transparency="OPAQUE",
|
|
classification="PUBLIC",
|
|
start_at=datetime(2026, 7, 8, 9 + index, tzinfo=timezone.utc),
|
|
all_day=False,
|
|
attendees=[],
|
|
categories=[],
|
|
rdate=[],
|
|
exdate=[],
|
|
reminders=[],
|
|
attachments=[],
|
|
related_to=[],
|
|
source_kind="ics",
|
|
source_href=f"ics-event-{index}",
|
|
icalendar={},
|
|
)
|
|
)
|
|
session.flush()
|
|
|
|
deleted = soft_delete_unseen_source_events(
|
|
session,
|
|
source=source,
|
|
seen_hrefs={"ics-event-1"},
|
|
batch_size=1,
|
|
)
|
|
|
|
active_hrefs = {
|
|
event.source_href
|
|
for event in session.query(CalendarEvent).filter(CalendarEvent.deleted_at.is_(None)).all()
|
|
}
|
|
self.assertEqual(deleted, 2)
|
|
self.assertEqual(active_hrefs, {"ics-event-1"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|