diff --git a/src/govoplan_identity/backend/directory.py b/src/govoplan_identity/backend/directory.py index ddb6508..416a11c 100644 --- a/src/govoplan_identity/backend/directory.py +++ b/src/govoplan_identity/backend/directory.py @@ -2,6 +2,8 @@ from __future__ import annotations from collections.abc import Sequence +from sqlalchemy import func, or_ + from govoplan_core.core.identity import IdentityAccountLinkRef, IdentityDirectory, IdentityRef from govoplan_core.db.session import get_database from govoplan_identity.backend.db.models import Identity, IdentityAccountLink @@ -103,3 +105,49 @@ class SqlIdentityDirectory(IdentityDirectory): .all() ) return tuple(_link_ref(link) for link in links) + + def search_identities( + self, + query: str | None = None, + *, + include_inactive: bool = False, + limit: int = 25, + ) -> tuple[IdentityRef, ...]: + normalized_limit = max(1, min(int(limit), 100)) + with get_database().session() as session: + identity_query = session.query(Identity) + if not include_inactive: + identity_query = identity_query.filter(Identity.is_active.is_(True)) + if query and query.strip(): + pattern = f"%{query.strip().casefold()}%" + matching_account_links = session.query(IdentityAccountLink.identity_id).filter( + func.lower(IdentityAccountLink.account_id).like(pattern) + ) + identity_query = identity_query.filter( + or_( + func.lower(Identity.id).like(pattern), + func.lower(Identity.display_name).like(pattern), + func.lower(Identity.external_subject).like(pattern), + Identity.id.in_(matching_account_links), + ) + ) + + identities = ( + identity_query + .order_by(Identity.display_name.asc(), Identity.id.asc()) + .limit(normalized_limit) + .all() + ) + identity_ids = [identity.id for identity in identities] + links = ( + session.query(IdentityAccountLink) + .filter(IdentityAccountLink.identity_id.in_(identity_ids)) + .order_by(IdentityAccountLink.is_primary.desc(), IdentityAccountLink.account_id.asc()) + .all() + if identity_ids + else [] + ) + links_by_identity: dict[str, list[IdentityAccountLink]] = {} + for link in links: + links_by_identity.setdefault(link.identity_id, []).append(link) + return tuple(_identity_ref(identity, links_by_identity.get(identity.id, ())) for identity in identities) diff --git a/src/govoplan_identity/backend/manifest.py b/src/govoplan_identity/backend/manifest.py index 934413a..d0977a0 100644 --- a/src/govoplan_identity/backend/manifest.py +++ b/src/govoplan_identity/backend/manifest.py @@ -3,7 +3,7 @@ from __future__ import annotations from pathlib import Path from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER -from govoplan_core.core.identity import CAPABILITY_IDENTITY_DIRECTORY +from govoplan_core.core.identity import CAPABILITY_IDENTITY_DIRECTORY, CAPABILITY_IDENTITY_SEARCH from govoplan_core.core.module_guards import persistent_table_uninstall_guard from govoplan_core.core.modules import DocumentationTopic, MigrationSpec, ModuleContext, ModuleManifest, PermissionDefinition, RoleTemplate from govoplan_core.db.base import Base @@ -75,6 +75,7 @@ manifest = ModuleManifest( ), capability_factories={ CAPABILITY_IDENTITY_DIRECTORY: _identity_directory, + CAPABILITY_IDENTITY_SEARCH: _identity_directory, }, documentation=( DocumentationTopic( diff --git a/tests/test_directory.py b/tests/test_directory.py new file mode 100644 index 0000000..0b1f120 --- /dev/null +++ b/tests/test_directory.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.identity import IdentitySearchProvider +from govoplan_core.db.base import Base +from govoplan_core.db.session import configure_database, reset_database +from govoplan_identity.backend.db.models import Identity, IdentityAccountLink +from govoplan_identity.backend.directory import SqlIdentityDirectory + + +class IdentityDirectoryTests(unittest.TestCase): + def setUp(self) -> None: + self.database = configure_database("sqlite:///:memory:") + Base.metadata.create_all( + self.database.engine, + tables=[Identity.__table__, IdentityAccountLink.__table__], + ) + with self.database.session() as session: + active = Identity( + id="identity-active", + display_name="Ada Example", + external_subject="subject-active", + source="local", + is_active=True, + settings={}, + ) + inactive = Identity( + id="identity-inactive", + display_name="Inactive Person", + external_subject="subject-inactive", + source="local", + is_active=False, + settings={}, + ) + session.add_all([active, inactive]) + session.flush() + session.add_all( + [ + IdentityAccountLink( + id="link-active", + identity_id=active.id, + account_id="account-ada", + is_primary=True, + source="local", + ), + IdentityAccountLink( + id="link-inactive", + identity_id=inactive.id, + account_id="account-inactive", + is_primary=True, + source="local", + ), + ] + ) + session.commit() + + def tearDown(self) -> None: + reset_database(dispose=True) + + def test_search_capability_filters_inactive_identities_and_searches_account_ids(self) -> None: + directory = SqlIdentityDirectory() + + self.assertIsInstance(directory, IdentitySearchProvider) + self.assertEqual( + ("identity-active",), + tuple(identity.id for identity in directory.search_identities()), + ) + matches = directory.search_identities("account-ada") + self.assertEqual(("identity-active",), tuple(identity.id for identity in matches)) + self.assertEqual(("account-ada",), matches[0].account_ids) + self.assertEqual("account-ada", matches[0].primary_account_id) + + def test_search_capability_can_include_inactive_identities(self) -> None: + directory = SqlIdentityDirectory() + + matches = directory.search_identities("inactive", include_inactive=True) + + self.assertEqual(("identity-inactive",), tuple(identity.id for identity in matches)) + + +if __name__ == "__main__": + unittest.main()