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.db.models import CalendarEvent from govoplan_calendar.backend.schemas import ( CalendarCollectionCreateRequest, CalendarEventCreateRequest, CalendarEventOccurrenceUpdateRequest, CalendarViewPreferencesUpdateRequest, ) from govoplan_calendar.backend.service import ( CalendarError, create_calendar, create_event, delete_event, delete_event_occurrence, get_calendar_view_preferences, list_event_occurrences, list_freebusy, update_calendar_view_preferences, update_event_occurrence, ) 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 CalendarRecurrenceAndPreferenceTests(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="Calendar"), ) self.session.flush() def tearDown(self) -> None: self.session.close() Base.metadata.drop_all(bind=self.engine) self.engine.dispose() def recurring_master(self) -> CalendarEvent: return create_event( self.session, tenant_id="tenant-1", user_id=None, payload=CalendarEventCreateRequest( calendar_id=self.calendar.id, uid="series@example.test", summary="Planning", start_at=datetime( 2026, 7, 8, 9, 0, tzinfo=timezone.utc ), end_at=datetime( 2026, 7, 8, 10, 0, tzinfo=timezone.utc ), rrule={"FREQ": "WEEKLY", "COUNT": "3"}, ), ) def test_occurrence_override_and_cancellation_reconcile_list_and_freebusy( self, ) -> None: master = self.recurring_master() override = update_event_occurrence( self.session, tenant_id="tenant-1", user_id=None, series_event_id=master.id, payload=CalendarEventOccurrenceUpdateRequest( recurrence_id="20260715T090000Z", summary="Moved planning", start_at=datetime( 2026, 7, 15, 13, 0, tzinfo=timezone.utc ), end_at=datetime( 2026, 7, 15, 14, 30, tzinfo=timezone.utc ), ), ) cancelled = delete_event_occurrence( self.session, tenant_id="tenant-1", user_id=None, series_event_id=master.id, recurrence_id="20260722T090000Z", ) self.session.commit() events = list_event_occurrences( self.session, tenant_id="tenant-1", start_at=datetime(2026, 7, 1, tzinfo=timezone.utc), end_at=datetime(2026, 7, 31, tzinfo=timezone.utc), ) self.assertEqual( [event["summary"] for event in events], ["Planning", "Moved planning"], ) self.assertEqual(events[1]["id"], override.id) self.assertEqual(events[1]["series_event_id"], master.id) self.assertTrue(events[1]["is_override"]) self.assertEqual( events[1]["start_at"], datetime(2026, 7, 15, 13, 0, tzinfo=timezone.utc), ) self.assertEqual(cancelled.status, "CANCELLED") busy = list_freebusy( self.session, tenant_id="tenant-1", start_at=datetime(2026, 7, 1, tzinfo=timezone.utc), end_at=datetime(2026, 7, 31, tzinfo=timezone.utc), ) self.assertEqual(len(busy), 2) self.assertEqual( busy[1]["start_at"], datetime(2026, 7, 15, 13, 0, tzinfo=timezone.utc), ) def test_deleting_series_removes_its_overrides(self) -> None: master = self.recurring_master() update_event_occurrence( self.session, tenant_id="tenant-1", user_id=None, series_event_id=master.id, payload=CalendarEventOccurrenceUpdateRequest( recurrence_id="20260715T090000Z", summary="Override", ), ) delete_event( self.session, tenant_id="tenant-1", event_id=master.id, ) active = ( self.session.query(CalendarEvent) .filter(CalendarEvent.deleted_at.is_(None)) .count() ) self.assertEqual(active, 0) def test_all_day_occurrence_retains_date_recurrence_id(self) -> None: master = create_event( self.session, tenant_id="tenant-1", user_id=None, payload=CalendarEventCreateRequest( calendar_id=self.calendar.id, uid="all-day-series@example.test", summary="All-day planning", start_at=datetime( 2026, 7, 8, tzinfo=timezone.utc ), end_at=datetime( 2026, 7, 9, tzinfo=timezone.utc ), all_day=True, rrule={"FREQ": "WEEKLY", "COUNT": "2"}, ), ) events = list_event_occurrences( self.session, tenant_id="tenant-1", start_at=datetime(2026, 7, 1, tzinfo=timezone.utc), end_at=datetime(2026, 7, 31, tzinfo=timezone.utc), ) self.assertEqual( [event["recurrence_id"] for event in events], ["20260708", "20260715"], ) override = update_event_occurrence( self.session, tenant_id="tenant-1", user_id=None, series_event_id=master.id, payload=CalendarEventOccurrenceUpdateRequest( recurrence_id=events[1]["recurrence_id"], summary="Moved all-day planning", ), ) self.assertEqual(override.recurrence_id, "20260715") def test_calendar_preferences_have_defaults_and_durable_user_overrides( self, ) -> None: defaults = get_calendar_view_preferences( self.session, tenant_id="tenant-1", user_id="user-1", ) self.assertTrue(defaults["dim_weekends"]) self.assertEqual(defaults["overridden_fields"], []) updated = update_calendar_view_preferences( self.session, tenant_id="tenant-1", user_id="user-1", payload=CalendarViewPreferencesUpdateRequest( dim_weekends=False, workday_start_hour=8, workday_end_hour=18, ), ) self.session.commit() reloaded = get_calendar_view_preferences( self.session, tenant_id="tenant-1", user_id="user-1", ) self.assertEqual(updated, reloaded) self.assertFalse(reloaded["dim_weekends"]) self.assertEqual(reloaded["workday_start_hour"], 8) self.assertEqual(reloaded["workday_end_hour"], 18) self.assertEqual( set(reloaded["overridden_fields"]), {"dim_weekends", "workday_start_hour", "workday_end_hour"}, ) def test_calendar_preferences_reject_inverted_workday(self) -> None: with self.assertRaisesRegex(CalendarError, "end hour"): update_calendar_view_preferences( self.session, tenant_id="tenant-1", user_id="user-1", payload=CalendarViewPreferencesUpdateRequest( workday_start_hour=18, workday_end_hour=8, ), ) if __name__ == "__main__": unittest.main()