96 lines
3.5 KiB
Python
96 lines
3.5 KiB
Python
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()
|