perf(access): provide bounded reference search

2026-07-30 01:30:04 +02:00
parent 6295cfa840
commit a758c8f2da
3 changed files with 430 additions and 0 deletions
--- a/tests/test_reference_options.py
+++ b/tests/test_reference_options.py
@@ -0,0 +1,153 @@
+from __future__ import annotations
+
+import unittest
+from types import SimpleNamespace
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_access.backend.db.models import Account, Group, User
+from govoplan_access.backend.reference_options import (
+    SqlAccessReferenceOptionProvider,
+)
+from govoplan_core.core.references import ReferenceSearchRequest
+from govoplan_core.db.base import Base
+
+
+class AccessReferenceOptionProviderTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite+pysqlite:///:memory:")
+        Base.metadata.create_all(
+            self.engine,
+            tables=[Account.__table__, User.__table__, Group.__table__],
+        )
+        self.session = Session(self.engine)
+        for index in range(120):
+            account = Account(
+                id=f"account-{index:03}",
+                email=f"person-{index:03}@example.test",
+                normalized_email=f"person-{index:03}@example.test",
+            )
+            self.session.add(account)
+            self.session.add(
+                User(
+                    id=f"membership-{index:03}",
+                    tenant_id="tenant-1",
+                    account_id=account.id,
+                    email=account.email,
+                    display_name=f"Person {index:03}",
+                )
+            )
+        for index in range(75):
+            self.session.add(
+                Group(
+                    id=f"group-{index:03}",
+                    tenant_id="tenant-1",
+                    slug=f"group-{index:03}",
+                    name=f"Group {index:03}",
+                )
+            )
+        self.session.commit()
+        self.provider = SqlAccessReferenceOptionProvider()
+        self.admin = SimpleNamespace(
+            tenant_id="tenant-1",
+            account_id="account-000",
+            group_ids=frozenset(),
+        )
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_large_directory_search_is_bounded_and_paged(self) -> None:
+        first = self.provider.search_reference_options(
+            self.session,
+            self.admin,
+            request=ReferenceSearchRequest(
+                kind="membership",
+                tenant_id="tenant-1",
+                limit=25,
+                context={"administrative": True},
+            ),
+        )
+        second = self.provider.search_reference_options(
+            self.session,
+            self.admin,
+            request=ReferenceSearchRequest(
+                kind="membership",
+                tenant_id="tenant-1",
+                limit=25,
+                cursor=first.next_cursor,
+                context={"administrative": True},
+            ),
+        )
+
+        self.assertEqual(25, len(first.options))
+        self.assertTrue(first.has_more)
+        self.assertEqual("offset:25", first.next_cursor)
+        self.assertEqual("membership-000", first.options[0].value)
+        self.assertEqual("membership-025", second.options[0].value)
+
+    def test_search_and_selected_values_do_not_materialize_the_directory(self) -> None:
+        page = self.provider.search_reference_options(
+            self.session,
+            self.admin,
+            request=ReferenceSearchRequest(
+                kind="membership",
+                tenant_id="tenant-1",
+                query="PERSON 119",
+                selected_values=("membership-005", "removed-membership"),
+                limit=10,
+                context={"administrative": True},
+            ),
+        )
+
+        self.assertEqual(
+            ["membership-119", "membership-005"],
+            [option.value for option in page.options],
+        )
+        self.assertFalse(page.has_more)
+
+    def test_non_administrators_only_search_their_permitted_references(self) -> None:
+        principal = SimpleNamespace(
+            tenant_id="tenant-1",
+            account_id="account-004",
+            group_ids=frozenset({"group-007"}),
+        )
+
+        users = self.provider.search_reference_options(
+            self.session,
+            principal,
+            request=ReferenceSearchRequest(
+                kind="user",
+                tenant_id="tenant-1",
+            ),
+        )
+        groups = self.provider.search_reference_options(
+            self.session,
+            principal,
+            request=ReferenceSearchRequest(
+                kind="group",
+                tenant_id="tenant-1",
+            ),
+        )
+
+        self.assertEqual(["account-004"], [option.value for option in users.options])
+        self.assertEqual(["group-007"], [option.value for option in groups.options])
+
+    def test_invalid_cursor_is_rejected(self) -> None:
+        with self.assertRaisesRegex(ValueError, "Invalid reference search cursor"):
+            self.provider.search_reference_options(
+                self.session,
+                self.admin,
+                request=ReferenceSearchRequest(
+                    kind="group",
+                    tenant_id="tenant-1",
+                    cursor="page:2",
+                    context={"administrative": True},
+                ),
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()