feat(access): expose tenant-bounded people search
This commit is contained in:
95
tests/test_people_search.py
Normal file
95
tests/test_people_search.py
Normal file
@@ -0,0 +1,95 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from types import SimpleNamespace
|
||||
import unittest
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from govoplan_access.backend.db.models import Account, User
|
||||
from govoplan_access.backend.manifest import manifest
|
||||
from govoplan_access.backend.people_search import AccessPeopleSearchProvider
|
||||
from govoplan_core.core.people import CAPABILITY_ACCESS_PEOPLE_SEARCH, PeopleSearchError, PeopleSearchProvider
|
||||
from govoplan_core.db.base import Base
|
||||
|
||||
|
||||
class AccessPeopleSearchTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.engine = create_engine("sqlite:///:memory:")
|
||||
Base.metadata.create_all(bind=self.engine)
|
||||
self.Session = sessionmaker(bind=self.engine)
|
||||
self.session = self.Session()
|
||||
self.provider = AccessPeopleSearchProvider()
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.session.close()
|
||||
Base.metadata.drop_all(bind=self.engine)
|
||||
self.engine.dispose()
|
||||
|
||||
def _add_user(
|
||||
self,
|
||||
suffix: str,
|
||||
*,
|
||||
tenant_id: str = "tenant-1",
|
||||
display_name: str | None = None,
|
||||
user_active: bool = True,
|
||||
account_active: bool = True,
|
||||
) -> None:
|
||||
email = f"{suffix}@example.test"
|
||||
account = Account(
|
||||
id=f"account-{suffix}",
|
||||
email=email,
|
||||
normalized_email=email,
|
||||
display_name=display_name,
|
||||
is_active=account_active,
|
||||
)
|
||||
self.session.add_all([
|
||||
account,
|
||||
User(
|
||||
id=f"user-{suffix}",
|
||||
tenant_id=tenant_id,
|
||||
account_id=account.id,
|
||||
email=email,
|
||||
display_name=display_name,
|
||||
is_active=user_active,
|
||||
),
|
||||
])
|
||||
self.session.flush()
|
||||
|
||||
def test_search_is_tenant_bounded_and_excludes_inactive_records(self) -> None:
|
||||
self._add_user("ada", display_name="Ada Lovelace")
|
||||
self._add_user("other", tenant_id="tenant-2", display_name="Ada Other Tenant")
|
||||
self._add_user("inactive-user", display_name="Ada Inactive User", user_active=False)
|
||||
self._add_user("inactive-account", display_name="Ada Inactive Account", account_active=False)
|
||||
|
||||
groups = self.provider.search_people(
|
||||
self.session,
|
||||
SimpleNamespace(tenant_id="tenant-1"),
|
||||
query="ada",
|
||||
)
|
||||
|
||||
self.assertEqual([group.key for group in groups], ["accounts"])
|
||||
self.assertEqual([item.reference_id for item in groups[0].candidates], ["account-ada"])
|
||||
self.assertEqual(groups[0].candidates[0].email, "ada@example.test")
|
||||
self.assertNotIn("tenant_id", groups[0].candidates[0].metadata)
|
||||
|
||||
def test_search_escapes_like_wildcards_and_requires_tenant_context(self) -> None:
|
||||
self._add_user("ada", display_name="Ada Lovelace")
|
||||
|
||||
groups = self.provider.search_people(
|
||||
self.session,
|
||||
SimpleNamespace(tenant_id="tenant-1"),
|
||||
query="%",
|
||||
)
|
||||
self.assertEqual(groups[0].candidates, ())
|
||||
with self.assertRaises(PeopleSearchError):
|
||||
self.provider.search_people(self.session, SimpleNamespace(tenant_id=None), query="ada")
|
||||
|
||||
def test_manifest_exposes_the_shared_interface(self) -> None:
|
||||
self.assertIn(CAPABILITY_ACCESS_PEOPLE_SEARCH, manifest.capability_factories)
|
||||
self.assertIn(CAPABILITY_ACCESS_PEOPLE_SEARCH, {item.name for item in manifest.provides_interfaces})
|
||||
self.assertIsInstance(self.provider, PeopleSearchProvider)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user