govoplan-calendar/tests/test_recurrence_preferences.py

from __future__ import annotations

import unittest
from datetime import datetime, timedelta, timezone
from unittest.mock import patch

from sqlalchemy import create_engine, create_mock_engine, event as sqlalchemy_event
from sqlalchemy.dialects import postgresql
from sqlalchemy.orm import Query, Session, 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,
    _bounded_occurrence_candidates,
    OCCURRENCE_DETAIL_FIELDS,
    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_freebusy_returns_all_1001_hourly_occurrences(self) -> None:
        start = datetime(2026, 7, 1, tzinfo=timezone.utc)
        create_event(
            self.session, tenant_id="tenant-1", user_id=None,
            payload=CalendarEventCreateRequest(
                calendar_id=self.calendar.id, uid="dense@example.test", summary="Dense series",
                start_at=start, end_at=start + timedelta(minutes=1),
                rrule={"FREQ": "HOURLY", "COUNT": "1001"},
            ),
        )
        self.session.commit()
        busy = list_freebusy(
            self.session, tenant_id="tenant-1", calendar_ids=[self.calendar.id],
            start_at=start, end_at=start + timedelta(days=43),
        )
        self.assertEqual(1001, len(busy))
        self.assertEqual(start + timedelta(hours=1000), busy[-1]["start_at"])

    def test_freebusy_over_budget_fails_without_partial_busy_slots(self) -> None:
        start = datetime(2026, 7, 1, tzinfo=timezone.utc)
        create_event(
            self.session, tenant_id="tenant-1", user_id=None,
            payload=CalendarEventCreateRequest(
                calendar_id=self.calendar.id, uid="excessive@example.test", summary="Excessive series",
                start_at=start, end_at=start + timedelta(seconds=1),
                rrule={"FREQ": "MINUTELY", "COUNT": "10001"},
            ),
        )
        self.session.commit()
        with self.assertRaisesRegex(CalendarError, "could not complete"):
            list_freebusy(
                self.session, tenant_id="tenant-1", calendar_ids=[self.calendar.id],
                start_at=start, end_at=start + timedelta(days=8),
            )
        self.assertEqual(1, self.session.query(CalendarEvent).count())

    def test_full_expansion_byte_cap_and_lightweight_complete_freebusy(self) -> None:
        from govoplan_calendar.backend import service

        start = datetime(2026, 7, 1, tzinfo=timezone.utc)
        create_event(
            self.session, tenant_id="tenant-1", user_id=None,
            payload=CalendarEventCreateRequest(
                calendar_id=self.calendar.id, uid="heavy@example.test", summary="Heavy series",
                description="x" * 65536, start_at=start, end_at=start + timedelta(minutes=1),
                rrule={"FREQ": "HOURLY", "COUNT": "1001"},
            ),
        )
        self.session.commit()
        calendar_id = self.calendar.id
        self.session.expunge_all()
        with patch.object(service, "expanded_event_response", wraps=service.expanded_event_response) as expand:
            with self.assertRaisesRegex(CalendarError, "response exceeds its byte limit"):
                list_event_occurrences(self.session, tenant_id="tenant-1", start_at=start, end_at=start + timedelta(days=43), limit=1)
            self.assertLess(expand.call_count, 70)
        statements = []
        loaded = []
        def capture(conn, cursor, statement, parameters, context, executemany):
            statements.append(statement)
        def capture_loaded(session, instance):
            if isinstance(instance, CalendarEvent):
                loaded.append(instance)
        sqlalchemy_event.listen(self.engine, "before_cursor_execute", capture)
        sqlalchemy_event.listen(self.session, "loaded_as_persistent", capture_loaded)
        try:
            with patch.object(service, "event_response", side_effect=AssertionError("Full event data is unnecessary for availability")):
                busy = list_freebusy(self.session, tenant_id="tenant-1", calendar_ids=[calendar_id], start_at=start, end_at=start + timedelta(days=43))
            self.assertEqual(1001, len(busy))
            self.assertEqual([], loaded)
            for field in ("description", "raw_ics", "icalendar", "metadata", "attendees", "attachments"):
                self.assertNotIn(f"calendar_events.{field}", "\n".join(statements))
        finally:
            sqlalchemy_event.remove(self.engine, "before_cursor_execute", capture)
            sqlalchemy_event.remove(self.session, "loaded_as_persistent", capture_loaded)

    def test_aggregate_sql_projection_hides_over_budget_values_before_driver_decoding(self) -> None:
        from govoplan_calendar.backend import service

        first = self.recurring_master()
        second = create_event(
            self.session, tenant_id="tenant-1", user_id=None,
            payload=CalendarEventCreateRequest(
                calendar_id=self.calendar.id, uid="second@example.test", summary="Second",
                start_at=first.start_at, end_at=first.end_at, rrule={"FREQ": "DAILY", "COUNT": "2"},
            ),
        )
        first.description = second.description = "x" * 1500
        self.session.commit()
        self.session.expunge_all()
        captured = []
        def capture(conn, cursor, statement, parameters, context, executemany):
            captured.append((statement, parameters))
        sqlalchemy_event.listen(self.engine, "before_cursor_execute", capture)
        try:
            with patch.object(service, "MAX_OCCURRENCE_PROJECTION_BYTES", 3000), patch.object(service, "expand_events_occurrences") as worker:
                with self.assertRaisesRegex(CalendarError, "candidate projection exceeds its byte limit"):
                    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),
                    )
                worker.assert_not_called()
        finally:
            sqlalchemy_event.remove(self.engine, "before_cursor_execute", capture)
        self.assertEqual(1, len(captured))
        statement, parameters = captured[0]
        self.assertIn("sum(", statement)
        self.assertIn("OVER (ORDER BY", statement)
        with self.engine.connect() as connection:
            raw = list(connection.exec_driver_sql(statement, parameters))
        self.assertEqual(2, len(raw))
        self.assertIsNotNone(raw[0][1])
        self.assertTrue(all(value is None for value in raw[1][1:]))

    def test_sql_projection_compiles_postgresql_without_binary_json_casts(self) -> None:
        statements = []
        session = Session(bind=create_mock_engine("postgresql://", lambda *args, **kwargs: None))
        def inspect_query(query):
            statements.append(str(query.statement.compile(dialect=postgresql.dialect())))
            return iter(())
        with patch.object(Query, "__iter__", inspect_query):
            self.assertEqual([], _bounded_occurrence_candidates(
                session.query(CalendarEvent).filter(CalendarEvent.tenant_id == "tenant-1").limit(2001),
                OCCURRENCE_DETAIL_FIELDS, [4096],
            ))
        self.assertIn("octet_length(CAST(calendar_events.icalendar AS TEXT))", statements[0])
        self.assertIn("CASE WHEN", statements[0])
        self.assertIn("OVER (ORDER BY", statements[0])
        self.assertNotIn("BYTEA", statements[0])

    def test_candidate_budget_is_not_bypassed_by_response_limit(self) -> None:
        self.recurring_master()
        self.session.commit()
        with patch("govoplan_calendar.backend.service.MAX_OCCURRENCE_CANDIDATES", 0):
            with self.assertRaisesRegex(CalendarError, "Too many recurring series"):
                list_event_occurrences(
                    self.session, tenant_id="tenant-1", limit=1,
                    start_at=datetime(2026, 7, 1, tzinfo=timezone.utc),
                    end_at=datetime(2026, 7, 31, tzinfo=timezone.utc),
                )

    def test_admission_rejection_precedes_database_projection(self) -> None:
        from govoplan_core.security.bounded_process import ProcessBudgetError

        with patch("govoplan_core.security.bounded_process.bounded_operation_admission", side_effect=ProcessBudgetError("busy")):
            with patch.object(self.session, "query") as query:
                with self.assertRaisesRegex(CalendarError, "could not complete"):
                    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),
                    )
                query.assert_not_called()

    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()