add-ideas/govoplan-core
1
0
Fork 0
Code Issues 45 Pull Requests Actions Packages Projects Releases Wiki Activity

Decouple scopes from tenancy schema
Some checks failed
Dependency Audit / dependency-audit (push) Has been cancelled
Details
Module Matrix / module-matrix (push) Has been cancelled
Details

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-10 17:33:43 +02:00
parent 635d25c74c
commit 79af252e88
27 changed files with 529 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
tests/test_identity_organization_contracts.py
View File

@@ -26,9 +26,15 @@ from govoplan_identity.backend.db.models import Identity, IdentityAccountLink
from govoplan_organizations.backend.db.models import (
OrganizationFunction,
OrganizationFunctionAssignment,
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
@@ -70,8 +76,8 @@ class _FakeOrganizationDirectory:
assignment = OrganizationFunctionAssignmentRef(
id="assignment-1",
tenant_id="tenant-1",
account_id="account-1",
identity_id="identity-1",
account_id="account-1",
function_id=function.id,
organization_unit_id=unit.id,
applies_to_subunits=True,
@@ -100,6 +106,13 @@ class _FakeOrganizationDirectory:
return ()
return (self.assignment,)
def 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:
@@ -129,6 +142,7 @@ class IdentityOrganizationContractTests(unittest.TestCase):
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={})
@@ -138,15 +152,60 @@ class IdentityOrganizationContractTests(unittest.TestCase):
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",
@@ -154,26 +213,44 @@ class IdentityOrganizationContractTests(unittest.TestCase):
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.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_organizations.backend.directory import SqlOrganizationDirectory
identity_directory = SqlIdentityDirectory()
organization_directory = SqlOrganizationDirectory()
organization_directory = SqlOrganizationDirectory(identity_directory=identity_directory)
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 organization_directory.function_assignments_for_identity("identity-directory", tenant_id="tenant-directory")],
)
self.assertEqual(
["assignment-directory"],
[item.id for item in organization_directory.function_assignments_for_account("account-directory", tenant_id="tenant-directory")],