From edb4687826e4b8e268e5b8165b92f1d6d24445a2 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Sat, 11 Jul 2026 01:13:53 +0200 Subject: [PATCH] Cover access canonical directory migration --- tests/test_identity_organization_contracts.py | 92 ++++++++++++++++++- 1 file changed, 90 insertions(+), 2 deletions(-) diff --git a/tests/test_identity_organization_contracts.py b/tests/test_identity_organization_contracts.py index 26489f3..2b5f837 100644 --- a/tests/test_identity_organization_contracts.py +++ b/tests/test_identity_organization_contracts.py @@ -18,6 +18,7 @@ from govoplan_core.core.idm import ( ) from govoplan_core.core.organizations import ( CAPABILITY_ORGANIZATION_DIRECTORY, + ORGANIZATIONS_MODULE_ID, OrganizationDirectory, OrganizationFunctionRef, OrganizationUnitRef, @@ -25,8 +26,8 @@ from govoplan_core.core.organizations import ( from govoplan_core.core.modules import ModuleContext, ModuleManifest from govoplan_core.core.registry import PlatformRegistry from govoplan_core.db.base import Base -from govoplan_access.backend.db.models import ExternalFunctionRoleAssignment, Role, User -from govoplan_access.backend.semantic import collect_external_function_roles +from govoplan_access.backend.db.models import Account, ExternalFunctionRoleAssignment, Role, User +from govoplan_access.backend.semantic import collect_external_function_roles, identity_id_for_account from govoplan_identity.backend.db.models import Identity, IdentityAccountLink from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment from govoplan_organizations.backend.db.models import ( @@ -93,6 +94,17 @@ class _FakeOrganizationDirectory: return (self.function,) if organization_unit_id == self.unit.id else () +class _InactiveOrganizationDirectory(_FakeOrganizationDirectory): + function = OrganizationFunctionRef( + id="function-1", + tenant_id="tenant-1", + organization_unit_id=_FakeOrganizationDirectory.unit.id, + slug="registry-clerk", + name="Registry Clerk", + status="inactive", + ) + + class _FakeIdmDirectory: assignment = OrganizationFunctionAssignmentRef( id="assignment-1", @@ -165,6 +177,74 @@ class IdentityOrganizationContractTests(unittest.TestCase): finally: shutil.rmtree(root, ignore_errors=True) + def test_access_directory_prefers_identity_directory_with_projection_fallback(self) -> None: + from govoplan_access.backend.directory import SqlAccessDirectory + + root = Path(tempfile.mkdtemp(prefix="govoplan-access-identity-directory-")) + try: + with temporary_database(f"sqlite:///{root / 'access-identity.db'}") as database: + Base.metadata.create_all(bind=database.engine) + with database.session() as session: + session.add( + Account( + id="account-1", + email="demo@example.test", + normalized_email="demo@example.test", + display_name="Demo", + ) + ) + session.commit() + + identity_directory = _FakeIdentityDirectory() + self.assertEqual( + "identity-1", + identity_id_for_account(session, "account-1", identity_directory=identity_directory), + ) + + directory = SqlAccessDirectory(identity_directory=_FakeIdentityDirectory()) + self.assertEqual("Demo Person", directory.get_identity("identity-1").display_name) # type: ignore[union-attr] + self.assertEqual(["account-1"], [item.id for item in directory.accounts_for_identity("identity-1")]) + + projection_only = SqlAccessDirectory() + self.assertIsNone(projection_only.get_identity("identity-1")) + finally: + shutil.rmtree(root, ignore_errors=True) + + def test_access_directory_prefers_organization_directory_for_functions(self) -> None: + from govoplan_access.backend.directory import SqlAccessDirectory + + root = Path(tempfile.mkdtemp(prefix="govoplan-access-organization-directory-")) + try: + with temporary_database(f"sqlite:///{root / 'access-organizations.db'}") as database: + Base.metadata.create_all(bind=database.engine) + with database.session() as session: + role = Role( + id="role-org-function", + tenant_id="tenant-1", + slug="org-function-reader", + name="Org Function Reader", + permissions=["files:file:read"], + is_assignable=True, + ) + mapping = ExternalFunctionRoleAssignment( + tenant_id="tenant-1", + source_module=ORGANIZATIONS_MODULE_ID, + function_id="function-1", + role_id=role.id, + settings={}, + ) + session.add_all([role, mapping]) + session.commit() + + directory = SqlAccessDirectory(organization_directory=_FakeOrganizationDirectory()) + self.assertEqual("Registry", directory.get_organization_unit("ou-1").name) # type: ignore[union-attr] + function = directory.get_function("function-1") + self.assertEqual("Registry Clerk", function.name) # type: ignore[union-attr] + self.assertEqual(("role-org-function",), function.role_ids) # type: ignore[union-attr] + self.assertEqual(["function-1"], [item.id for item in directory.functions_for_organization_unit("ou-1")]) + finally: + shutil.rmtree(root, ignore_errors=True) + def test_access_maps_idm_function_facts_to_roles_explicitly(self) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-access-idm-role-map-")) try: @@ -202,8 +282,16 @@ class IdentityOrganizationContractTests(unittest.TestCase): session, user, _FakeIdmDirectory().organization_function_assignments_for_account("account-1", tenant_id="tenant-1"), + organization_directory=_FakeOrganizationDirectory(), ) self.assertEqual(["role-idm-function"], [item.id for item in roles]) + inactive_roles = collect_external_function_roles( + session, + user, + _FakeIdmDirectory().organization_function_assignments_for_account("account-1", tenant_id="tenant-1"), + organization_directory=_InactiveOrganizationDirectory(), + ) + self.assertEqual([], inactive_roles) finally: shutil.rmtree(root, ignore_errors=True)