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 = """ EWS item 2026-07-08T09:00:00Z 2026-07-08T10:00:00Z false ews-uid-1 """ 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(" 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()