187 lines
8.3 KiB
Python
187 lines
8.3 KiB
Python
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.organizations import (
|
|
CAPABILITY_ORGANIZATION_DIRECTORY,
|
|
OrganizationDirectory,
|
|
OrganizationFunctionAssignmentRef,
|
|
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 User # noqa: F401 - registers legacy tenancy relationship target
|
|
from govoplan_identity.backend.db.models import Identity, IdentityAccountLink
|
|
from govoplan_organizations.backend.db.models import (
|
|
OrganizationFunction,
|
|
OrganizationFunctionAssignment,
|
|
OrganizationUnit,
|
|
)
|
|
from govoplan_tenancy.backend.db.models import Tenant
|
|
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",
|
|
)
|
|
assignment = OrganizationFunctionAssignmentRef(
|
|
id="assignment-1",
|
|
tenant_id="tenant-1",
|
|
account_id="account-1",
|
|
identity_id="identity-1",
|
|
function_id=function.id,
|
|
organization_unit_id=unit.id,
|
|
applies_to_subunits=True,
|
|
)
|
|
|
|
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 ()
|
|
|
|
def get_function_assignment(self, assignment_id: str) -> OrganizationFunctionAssignmentRef | None:
|
|
return self.assignment if assignment_id == self.assignment.id else None
|
|
|
|
def 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,)
|
|
|
|
|
|
class IdentityOrganizationContractTests(unittest.TestCase):
|
|
def test_protocol_shapes_match_reference_implementations(self) -> None:
|
|
self.assertIsInstance(_FakeIdentityDirectory(), IdentityDirectory)
|
|
self.assertIsInstance(_FakeOrganizationDirectory(), OrganizationDirectory)
|
|
|
|
def test_capabilities_register_and_resolve_through_platform_registry(self) -> None:
|
|
identity_directory = _FakeIdentityDirectory()
|
|
organization_directory = _FakeOrganizationDirectory()
|
|
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,
|
|
},
|
|
)
|
|
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))
|
|
|
|
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:
|
|
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 = OrganizationUnit(
|
|
id="ou-directory",
|
|
tenant_id=tenant.id,
|
|
slug="registry",
|
|
name="Registry",
|
|
)
|
|
function = OrganizationFunction(
|
|
id="function-directory",
|
|
tenant_id=tenant.id,
|
|
organization_unit_id=unit.id,
|
|
slug="registry-clerk",
|
|
name="Registry Clerk",
|
|
)
|
|
assignment = OrganizationFunctionAssignment(
|
|
id="assignment-directory",
|
|
tenant_id=tenant.id,
|
|
account_id="account-directory",
|
|
identity_id=identity.id,
|
|
function_id=function.id,
|
|
organization_unit_id=unit.id,
|
|
applies_to_subunits=True,
|
|
)
|
|
session.add_all([tenant, identity, link, unit, function, assignment])
|
|
session.commit()
|
|
|
|
from govoplan_identity.backend.directory import SqlIdentityDirectory
|
|
from govoplan_organizations.backend.directory import SqlOrganizationDirectory
|
|
|
|
identity_directory = SqlIdentityDirectory()
|
|
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("Registry Clerk", organization_directory.get_function("function-directory").name) # type: ignore[union-attr]
|
|
self.assertEqual(
|
|
["assignment-directory"],
|
|
[item.id for item in organization_directory.function_assignments_for_account("account-directory", tenant_id="tenant-directory")],
|
|
)
|
|
finally:
|
|
shutil.rmtree(root, ignore_errors=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|