Add native permission-aware IDM search source

2026-08-04 03:03:27 +02:00
parent d8643174d7
commit ad37b030e2
3 changed files with 519 additions and 0 deletions
@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.events import EventObjectRef, EventTenantRef, PlatformEvent
+from govoplan_core.core.search import (
+    SearchAuthorizationRequest,
+    SearchBackfillRequest,
+    SearchResourceReference,
+)
+from govoplan_core.db.base import Base
+from govoplan_idm.backend.db.models import IdmTypedGroup
+from govoplan_idm.backend.search_source import IdmSearchSource, PROVIDER_ID
+
+
+class IdmSearchSourceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite://")
+        Base.metadata.create_all(
+            self.engine,
+            tables=(IdmTypedGroup.__table__,),
+        )
+        self.session = Session(self.engine)
+        self.session.add_all(
+            (
+                IdmTypedGroup(
+                    id="group-1",
+                    tenant_id="tenant-1",
+                    key="permit-holder",
+                    name="Permit holders",
+                    group_type="business_status",
+                ),
+                IdmTypedGroup(
+                    id="group-other",
+                    tenant_id="tenant-2",
+                    key="other",
+                    name="Other tenant",
+                    group_type="business_status",
+                ),
+            )
+        )
+        self.session.commit()
+        self.source = IdmSearchSource()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_typed_group_search_is_tenant_and_scope_bounded(self) -> None:
+        page = self.source.backfill(
+            self.session,
+            request=SearchBackfillRequest(
+                tenant_id="tenant-1",
+                provider_id=PROVIDER_ID,
+                resource_type="typed_group",
+                rebuild_id="rebuild-1",
+            ),
+        )
+        self.assertEqual(("group-1",), tuple(doc.resource_id for doc in page.documents))
+        reference = SearchResourceReference(
+            tenant_id="tenant-1",
+            module_id="idm",
+            resource_type="typed_group",
+            resource_id="group-1",
+        )
+        request = SearchAuthorizationRequest(reference=reference, source_revision="1")
+        self.assertTrue(
+            self.source.authorize(
+                self.session,
+                _principal({"idm:relationship:read"}),
+                requests=(request,),
+            )[reference.key]
+        )
+        self.assertFalse(
+            self.source.authorize(
+                self.session,
+                _principal(set()),
+                requests=(request,),
+            )[reference.key]
+        )
+
+    def test_typed_group_event_is_translated_to_an_upsert(self) -> None:
+        event = PlatformEvent(
+            type="idm.typed_group.updated",
+            module_id="idm",
+            tenant=EventTenantRef(id="tenant-1"),
+            resource=EventObjectRef(type="typed_group", id="group-1"),
+        )
+        change = self.source.index_changes_for_event(
+            self.session,
+            event=event,
+            delivery_key="delivery-1",
+        )[0]
+        self.assertEqual("upsert", change.kind)
+        self.assertEqual("group-1", change.reference.resource_id)
+
+
+def _principal(scopes: set[str]) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="user-1",
+            tenant_id="tenant-1",
+            scopes=frozenset(scopes),
+        ),
+        account=SimpleNamespace(id="account-1"),
+        user=SimpleNamespace(id="user-1"),
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()