Add native permission-aware calendar search source

2026-08-04 03:03:26 +02:00
parent d7fd9447a2
commit 04438d96ca
3 changed files with 450 additions and 1 deletions
@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_access.backend.db.models import Account, User
+from govoplan_calendar.backend.db.models import CalendarCollection, CalendarEvent
+from govoplan_calendar.backend.search_source import (
+    CalendarSearchSource,
+    PROVIDER_ID,
+    RESOURCE_TYPE,
+)
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.search import (
+    SearchAuthorizationRequest,
+    SearchBackfillRequest,
+    SearchResourceReference,
+)
+from govoplan_core.db.base import Base
+
+
+class CalendarSearchSourceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite://")
+        Base.metadata.create_all(
+            self.engine,
+            tables=(
+                Account.__table__,
+                User.__table__,
+                CalendarCollection.__table__,
+                CalendarEvent.__table__,
+            ),
+        )
+        self.session = Session(self.engine)
+        start = datetime(2026, 8, 5, 9, tzinfo=timezone.utc)
+        self.session.add_all(
+            (
+                Account(
+                    id="account-1",
+                    email="one@example.test",
+                    normalized_email="one@example.test",
+                ),
+                User(
+                    id="user-1",
+                    tenant_id="tenant-1",
+                    account_id="account-1",
+                    email="one@example.test",
+                ),
+                CalendarCollection(
+                    id="calendar-1",
+                    tenant_id="tenant-1",
+                    slug="private",
+                    name="Private calendar",
+                    owner_type="user",
+                    owner_id="user-1",
+                    visibility="private",
+                    created_by_user_id="user-1",
+                ),
+                CalendarEvent(
+                    id="event-1",
+                    tenant_id="tenant-1",
+                    calendar_id="calendar-1",
+                    uid="event-1@example.test",
+                    summary="Permit review",
+                    description="Review monthly permits",
+                    start_at=start,
+                    end_at=start + timedelta(hours=1),
+                ),
+            )
+        )
+        self.session.commit()
+        self.source = CalendarSearchSource()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_private_calendar_is_visible_only_to_current_owner(self) -> None:
+        page = self.source.backfill(
+            self.session,
+            request=SearchBackfillRequest(
+                tenant_id="tenant-1",
+                provider_id=PROVIDER_ID,
+                resource_type=RESOURCE_TYPE,
+                rebuild_id="rebuild-1",
+            ),
+        )
+        self.assertEqual(("event-1",), tuple(doc.resource_id for doc in page.documents))
+        reference = SearchResourceReference(
+            tenant_id="tenant-1",
+            module_id="calendar",
+            resource_type=RESOURCE_TYPE,
+            resource_id="event-1",
+        )
+        request = SearchAuthorizationRequest(reference=reference, source_revision="1")
+        self.assertTrue(
+            self.source.authorize(
+                self.session,
+                _principal("user-1"),
+                requests=(request,),
+            )[reference.key]
+        )
+        self.assertFalse(
+            self.source.authorize(
+                self.session,
+                _principal("user-2"),
+                requests=(request,),
+            )[reference.key]
+        )
+
+
+def _principal(user_id: str) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id=f"account-{user_id}",
+            membership_id=user_id,
+            tenant_id="tenant-1",
+            scopes=frozenset({"calendar:event:read"}),
+        ),
+        account=SimpleNamespace(id=f"account-{user_id}"),
+        user=SimpleNamespace(id=user_id),
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()