Cover access canonical directory migration

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.db.models import Account, ExternalFunctionRoleAssignment, Role, User
-from govoplan_access.backend.semantic import collect_external_function_roles
+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)