from __future__ import annotations import shutil import tempfile import unittest from pathlib import Path from govoplan_core.core.identity import ( CAPABILITY_IDENTITY_DIRECTORY, IdentityAccountLinkRef, IdentityDirectory, IdentityRef, ) from govoplan_core.core.idm import ( CAPABILITY_IDM_DIRECTORY, IdmDirectory, OrganizationFunctionAssignmentRef, ) from govoplan_core.core.organizations import ( CAPABILITY_ORGANIZATION_DIRECTORY, OrganizationDirectory, OrganizationFunctionRef, OrganizationUnitRef, ) 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_identity.backend.db.models import Identity, IdentityAccountLink from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment from govoplan_organizations.backend.db.models import ( OrganizationFunction, OrganizationFunctionType, OrganizationRelation, OrganizationRelationType, OrganizationStructure, OrganizationUnit, OrganizationUnitType, ) from govoplan_tenancy.backend.db.models import Tenant from govoplan_core.tenancy.scope import create_scope_tables from tests.db_isolation import temporary_database class _FakeIdentityDirectory: identity = IdentityRef( id="identity-1", display_name="Demo Person", primary_account_id="account-1", account_ids=("account-1", "account-2"), ) def get_identity(self, identity_id: str) -> IdentityRef | None: return self.identity if identity_id == self.identity.id else None def identity_for_account(self, account_id: str) -> IdentityRef | None: return self.identity if account_id in self.identity.account_ids else None def identities_for_accounts(self, account_ids): return (self.identity,) if set(account_ids) & set(self.identity.account_ids) else () def accounts_for_identity(self, identity_id: str): if identity_id != self.identity.id: return () return ( IdentityAccountLinkRef(id="link-1", identity_id=identity_id, account_id="account-1", is_primary=True), IdentityAccountLinkRef(id="link-2", identity_id=identity_id, account_id="account-2"), ) class _FakeOrganizationDirectory: unit = OrganizationUnitRef(id="ou-1", tenant_id="tenant-1", slug="registry", name="Registry") function = OrganizationFunctionRef( id="function-1", tenant_id="tenant-1", organization_unit_id=unit.id, slug="registry-clerk", name="Registry Clerk", ) def get_organization_unit(self, organization_unit_id: str) -> OrganizationUnitRef | None: return self.unit if organization_unit_id == self.unit.id else None def organization_units_for_tenant(self, tenant_id: str): return (self.unit,) if tenant_id == self.unit.tenant_id else () def get_function(self, function_id: str) -> OrganizationFunctionRef | None: return self.function if function_id == self.function.id else None def functions_for_organization_unit(self, organization_unit_id: str, *, include_subunits: bool = False): del include_subunits return (self.function,) if organization_unit_id == self.unit.id else () class _FakeIdmDirectory: assignment = OrganizationFunctionAssignmentRef( id="assignment-1", tenant_id="tenant-1", identity_id="identity-1", account_id="account-1", function_id=_FakeOrganizationDirectory.function.id, organization_unit_id=_FakeOrganizationDirectory.unit.id, applies_to_subunits=True, ) def get_organization_function_assignment(self, assignment_id: str) -> OrganizationFunctionAssignmentRef | None: return self.assignment if assignment_id == self.assignment.id else None def organization_function_assignments_for_account(self, account_id: str, *, tenant_id: str | None = None): if account_id != self.assignment.account_id: return () if tenant_id is not None and tenant_id != self.assignment.tenant_id: return () return (self.assignment,) def organization_function_assignments_for_identity(self, identity_id: str, *, tenant_id: str | None = None): if identity_id != self.assignment.identity_id: return () if tenant_id is not None and tenant_id != self.assignment.tenant_id: return () return (self.assignment,) class IdentityOrganizationContractTests(unittest.TestCase): def test_protocol_shapes_match_reference_implementations(self) -> None: self.assertIsInstance(_FakeIdentityDirectory(), IdentityDirectory) self.assertIsInstance(_FakeOrganizationDirectory(), OrganizationDirectory) self.assertIsInstance(_FakeIdmDirectory(), IdmDirectory) def test_capabilities_register_and_resolve_through_platform_registry(self) -> None: identity_directory = _FakeIdentityDirectory() organization_directory = _FakeOrganizationDirectory() idm_directory = _FakeIdmDirectory() manifest = ModuleManifest( id="directory-contract-test", name="Directory Contract Test", version="test", capability_factories={ CAPABILITY_IDENTITY_DIRECTORY: lambda context: identity_directory, CAPABILITY_ORGANIZATION_DIRECTORY: lambda context: organization_directory, CAPABILITY_IDM_DIRECTORY: lambda context: idm_directory, }, ) registry = PlatformRegistry() registry.register(manifest) registry.configure_capability_context(ModuleContext(registry=registry, settings=object())) self.assertIs(identity_directory, registry.require_capability(CAPABILITY_IDENTITY_DIRECTORY)) self.assertIs(organization_directory, registry.require_capability(CAPABILITY_ORGANIZATION_DIRECTORY)) self.assertIs(idm_directory, registry.require_capability(CAPABILITY_IDM_DIRECTORY)) def test_access_directory_can_optionally_consume_idm_assignments(self) -> None: from govoplan_access.backend.directory import SqlAccessDirectory root = Path(tempfile.mkdtemp(prefix="govoplan-access-idm-directory-")) try: with temporary_database(f"sqlite:///{root / 'access-idm.db'}") as database: Base.metadata.create_all(bind=database.engine) directory = SqlAccessDirectory(idm_directory=_FakeIdmDirectory()) assignments = directory.function_assignments_for_account("account-1", tenant_id="tenant-1") self.assertEqual(("assignment-1",), tuple(item.id for item in assignments)) self.assertEqual("identity-1", assignments[0].identity_id) self.assertEqual("function-1", assignments[0].function_id) 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: with temporary_database(f"sqlite:///{root / 'access-idm-role-map.db'}") as database: Base.metadata.create_all(bind=database.engine) with database.session() as session: user = User( id="user-idm-role-map", tenant_id="tenant-1", account_id="account-1", email="role-map@example.test", display_name="Role Map", settings={}, mail_profile_policy={}, ) role = Role( id="role-idm-function", tenant_id="tenant-1", slug="idm-function-reader", name="IDM Function Reader", permissions=["files:file:read"], is_assignable=True, ) mapping = ExternalFunctionRoleAssignment( tenant_id="tenant-1", source_module="organizations", function_id="function-1", role_id=role.id, settings={}, ) session.add_all([user, role, mapping]) session.commit() roles = collect_external_function_roles( session, user, _FakeIdmDirectory().organization_function_assignments_for_account("account-1", tenant_id="tenant-1"), ) self.assertEqual(["role-idm-function"], [item.id for item in roles]) finally: shutil.rmtree(root, ignore_errors=True) def test_sql_identity_and_organization_directories_return_dtos(self) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-directory-contracts-")) try: with temporary_database(f"sqlite:///{root / 'directory.db'}") as database: create_scope_tables(database.engine) Base.metadata.create_all(bind=database.engine) with database.session() as session: tenant = Tenant(id="tenant-directory", slug="directory", name="Directory Tenant", settings={}) identity = Identity(id="identity-directory", display_name="Directory Person", source="local") link = IdentityAccountLink( identity_id=identity.id, account_id="account-directory", is_primary=True, ) unit_type = OrganizationUnitType( id="unit-type-directory", tenant_id=tenant.id, slug="office", name="Office", ) structure = OrganizationStructure( id="structure-directory", tenant_id=tenant.id, slug="employer", name="Employer hierarchy", ) relation_type = OrganizationRelationType( id="relation-type-directory", tenant_id=tenant.id, structure_id=structure.id, slug="supervises", name="Supervises", source_unit_type_id=unit_type.id, target_unit_type_id=unit_type.id, ) unit = OrganizationUnit( id="ou-directory", tenant_id=tenant.id, unit_type_id=unit_type.id, slug="registry", name="Registry", ) child_unit = OrganizationUnit( id="ou-directory-child", tenant_id=tenant.id, unit_type_id=unit_type.id, slug="registry-frontdesk", name="Registry Frontdesk", ) relation = OrganizationRelation( id="relation-directory", tenant_id=tenant.id, structure_id=structure.id, relation_type_id=relation_type.id, source_unit_id=unit.id, target_unit_id=child_unit.id, ) function_type = OrganizationFunctionType( id="function-type-directory", tenant_id=tenant.id, slug="clerk", name="Clerk", organization_unit_type_id=unit_type.id, ) function = OrganizationFunction( id="function-directory", tenant_id=tenant.id, function_type_id=function_type.id, organization_unit_id=unit.id, slug="registry-clerk", name="Registry Clerk", ) assignment = IdmOrganizationFunctionAssignment( id="assignment-directory", tenant_id=tenant.id, identity_id=identity.id, function_id=function.id, organization_unit_id=unit.id, applies_to_subunits=True, ) session.add_all([ tenant, identity, link, unit_type, structure, relation_type, unit, child_unit, relation, function_type, function, assignment, ]) session.commit() from govoplan_identity.backend.directory import SqlIdentityDirectory from govoplan_idm.backend.directory import SqlIdmDirectory from govoplan_organizations.backend.directory import SqlOrganizationDirectory identity_directory = SqlIdentityDirectory() idm_directory = SqlIdmDirectory() organization_directory = SqlOrganizationDirectory() self.assertEqual("Directory Person", identity_directory.get_identity("identity-directory").display_name) # type: ignore[union-attr] self.assertEqual("identity-directory", identity_directory.identity_for_account("account-directory").id) # type: ignore[union-attr] self.assertEqual(["account-directory"], [item.account_id for item in identity_directory.accounts_for_identity("identity-directory")]) self.assertEqual("Registry", organization_directory.get_organization_unit("ou-directory").name) # type: ignore[union-attr] self.assertEqual("unit-type-directory", organization_directory.get_organization_unit("ou-directory").unit_type_id) # type: ignore[union-attr] self.assertEqual("Registry Clerk", organization_directory.get_function("function-directory").name) # type: ignore[union-attr] self.assertEqual("function-type-directory", organization_directory.get_function("function-directory").function_type_id) # type: ignore[union-attr] self.assertEqual( ["assignment-directory"], [item.id for item in idm_directory.organization_function_assignments_for_identity("identity-directory", tenant_id="tenant-directory")], ) self.assertEqual( ["assignment-directory"], [item.id for item in idm_directory.organization_function_assignments_for_account("account-directory", tenant_id="tenant-directory")], ) finally: shutil.rmtree(root, ignore_errors=True) if __name__ == "__main__": unittest.main()