337 lines
12 KiB
Python
337 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from govoplan_core.core.identity import IdentityAccountLinkRef, IdentityRef
|
|
from govoplan_core.core.organizations import OrganizationFunctionRef
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.db.session import configure_database, reset_database
|
|
from govoplan_identity.backend.db import models as identity_models # noqa: F401 - resolve assignment foreign keys
|
|
from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment
|
|
from govoplan_idm.backend.directory import SqlIdmDirectory
|
|
from govoplan_organizations.backend.db import models as organization_models # noqa: F401 - resolve assignment foreign keys
|
|
|
|
|
|
class StubIdentityDirectory:
|
|
def __init__(self, identities: tuple[IdentityRef, ...] = ()) -> None:
|
|
self.identities = identities
|
|
|
|
def get_identity(self, identity_id: str) -> IdentityRef | None:
|
|
return next(
|
|
(item for item in self.identities if item.id == identity_id),
|
|
None,
|
|
)
|
|
|
|
def identity_for_account(self, account_id: str) -> IdentityRef | None:
|
|
return next(
|
|
(
|
|
item
|
|
for item in self.identities
|
|
if account_id in item.account_ids
|
|
or item.primary_account_id == account_id
|
|
),
|
|
None,
|
|
)
|
|
|
|
def identities_for_accounts(self, account_ids: tuple[str, ...]) -> tuple[IdentityRef, ...]:
|
|
requested = set(account_ids)
|
|
return tuple(
|
|
item
|
|
for item in self.identities
|
|
if requested.intersection(item.account_ids)
|
|
or item.primary_account_id in requested
|
|
)
|
|
|
|
def accounts_for_identity(self, identity_id: str) -> tuple[IdentityAccountLinkRef, ...]:
|
|
identity = self.get_identity(identity_id)
|
|
if identity is None:
|
|
return ()
|
|
return tuple(
|
|
IdentityAccountLinkRef(
|
|
id=f"{identity_id}:{account_id}",
|
|
identity_id=identity_id,
|
|
account_id=account_id,
|
|
is_primary=account_id == identity.primary_account_id,
|
|
)
|
|
for account_id in identity.account_ids
|
|
)
|
|
|
|
|
|
class StubOrganizationDirectory:
|
|
def get_function(self, function_id: str) -> OrganizationFunctionRef | None:
|
|
return OrganizationFunctionRef(
|
|
id=function_id,
|
|
tenant_id="tenant-1",
|
|
organization_unit_id="unit-1",
|
|
slug=function_id,
|
|
name=function_id,
|
|
status="active",
|
|
)
|
|
|
|
def get_organization_unit(self, organization_unit_id: str):
|
|
return None
|
|
|
|
def organization_units_for_tenant(self, tenant_id: str):
|
|
return ()
|
|
|
|
def functions_for_organization_unit(self, organization_unit_id: str, *, include_subunits: bool = False):
|
|
return ()
|
|
|
|
|
|
class IdmDirectoryDelegationTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.database = configure_database("sqlite:///:memory:")
|
|
Base.metadata.create_all(
|
|
self.database.engine,
|
|
tables=[IdmOrganizationFunctionAssignment.__table__],
|
|
)
|
|
self.directory = SqlIdmDirectory(
|
|
identities=StubIdentityDirectory(), # type: ignore[arg-type]
|
|
organizations=StubOrganizationDirectory(), # type: ignore[arg-type]
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
reset_database(dispose=True)
|
|
|
|
def _source_and_child(
|
|
self,
|
|
*,
|
|
case: str,
|
|
child_source: str = "delegated",
|
|
source_tenant_id: str = "tenant-1",
|
|
source_function_id: str = "function-1",
|
|
source_is_active: bool = True,
|
|
source_valid_from: datetime | None = None,
|
|
source_valid_until: datetime | None = None,
|
|
) -> tuple[IdmOrganizationFunctionAssignment, IdmOrganizationFunctionAssignment]:
|
|
source = IdmOrganizationFunctionAssignment(
|
|
id=f"source-{case}",
|
|
tenant_id=source_tenant_id,
|
|
identity_id=f"source-identity-{case}",
|
|
function_id=source_function_id,
|
|
organization_unit_id="unit-1",
|
|
source="direct",
|
|
is_active=source_is_active,
|
|
valid_from=source_valid_from,
|
|
valid_until=source_valid_until,
|
|
settings={},
|
|
)
|
|
child = IdmOrganizationFunctionAssignment(
|
|
id=f"child-{case}",
|
|
tenant_id="tenant-1",
|
|
identity_id=f"target-identity-{case}",
|
|
function_id="function-1",
|
|
organization_unit_id="unit-1",
|
|
source=child_source,
|
|
delegated_from_assignment_id=source.id,
|
|
acting_for_account_id="represented-account" if child_source == "acting_for" else None,
|
|
is_active=True,
|
|
settings={},
|
|
)
|
|
return source, child
|
|
|
|
def test_effective_delegations_require_a_current_matching_source(self) -> None:
|
|
now = datetime.now(timezone.utc)
|
|
cases = (
|
|
self._source_and_child(case="valid"),
|
|
self._source_and_child(case="acting", child_source="acting_for"),
|
|
self._source_and_child(case="revoked", source_is_active=False),
|
|
self._source_and_child(case="expired", source_valid_until=now - timedelta(days=1)),
|
|
self._source_and_child(case="future", source_valid_from=now + timedelta(days=1)),
|
|
self._source_and_child(case="wrong-tenant", source_tenant_id="tenant-2"),
|
|
self._source_and_child(case="wrong-function", source_function_id="function-2"),
|
|
)
|
|
with self.database.session() as session:
|
|
for source, child in cases:
|
|
session.add_all((source, child))
|
|
session.commit()
|
|
|
|
expected = {
|
|
"valid": ("child-valid",),
|
|
"acting": ("child-acting",),
|
|
"revoked": (),
|
|
"expired": (),
|
|
"future": (),
|
|
"wrong-tenant": (),
|
|
"wrong-function": (),
|
|
}
|
|
for case, expected_ids in expected.items():
|
|
with self.subTest(case=case):
|
|
assignments = self.directory.organization_function_assignments_for_identity(
|
|
f"target-identity-{case}",
|
|
tenant_id="tenant-1",
|
|
)
|
|
self.assertEqual(expected_ids, tuple(item.id for item in assignments))
|
|
|
|
def test_reverse_lookup_returns_only_effective_function_assignments(self) -> None:
|
|
now = datetime.now(timezone.utc)
|
|
active = IdmOrganizationFunctionAssignment(
|
|
id="active-function-holder",
|
|
tenant_id="tenant-1",
|
|
identity_id="identity-active",
|
|
account_id="account-active",
|
|
function_id="function-1",
|
|
organization_unit_id="unit-1",
|
|
source="direct",
|
|
is_active=True,
|
|
settings={},
|
|
)
|
|
expired = IdmOrganizationFunctionAssignment(
|
|
id="expired-function-holder",
|
|
tenant_id="tenant-1",
|
|
identity_id="identity-expired",
|
|
function_id="function-1",
|
|
organization_unit_id="unit-1",
|
|
source="direct",
|
|
is_active=True,
|
|
valid_until=now - timedelta(minutes=1),
|
|
settings={},
|
|
)
|
|
other_tenant = IdmOrganizationFunctionAssignment(
|
|
id="other-tenant-holder",
|
|
tenant_id="tenant-2",
|
|
identity_id="identity-other",
|
|
function_id="function-1",
|
|
organization_unit_id="unit-1",
|
|
source="direct",
|
|
is_active=True,
|
|
settings={},
|
|
)
|
|
with self.database.session() as session:
|
|
session.add_all((active, expired, other_tenant))
|
|
session.commit()
|
|
|
|
assignments = self.directory.organization_function_assignments_for_function(
|
|
"function-1",
|
|
tenant_id="tenant-1",
|
|
)
|
|
|
|
self.assertEqual(
|
|
("active-function-holder",),
|
|
tuple(item.id for item in assignments),
|
|
)
|
|
|
|
def test_batch_incumbency_reports_vacancy_and_honors_effective_time(
|
|
self,
|
|
) -> None:
|
|
boundary = datetime.now(timezone.utc)
|
|
assignment = IdmOrganizationFunctionAssignment(
|
|
id="bounded-holder",
|
|
tenant_id="tenant-1",
|
|
identity_id="identity-bounded",
|
|
account_id="account-bounded",
|
|
function_id="function-1",
|
|
organization_unit_id="unit-1",
|
|
source="direct",
|
|
is_active=True,
|
|
valid_from=boundary - timedelta(hours=1),
|
|
valid_until=boundary + timedelta(hours=1),
|
|
settings={},
|
|
)
|
|
with self.database.session() as session:
|
|
session.add(assignment)
|
|
session.commit()
|
|
|
|
current = self.directory.organization_function_incumbencies(
|
|
("function-1", "function-2"),
|
|
tenant_id="tenant-1",
|
|
effective_at=boundary,
|
|
)
|
|
later = self.directory.organization_function_incumbencies(
|
|
("function-1",),
|
|
tenant_id="tenant-1",
|
|
effective_at=boundary + timedelta(hours=2),
|
|
)
|
|
|
|
self.assertEqual(
|
|
("bounded-holder",),
|
|
tuple(item.id for item in current["function-1"].assignments),
|
|
)
|
|
self.assertFalse(current["function-1"].vacant)
|
|
self.assertTrue(current["function-2"].vacant)
|
|
self.assertTrue(later["function-1"].vacant)
|
|
|
|
with self.assertRaisesRegex(ValueError, "another tenant"):
|
|
self.directory.organization_function_incumbencies(
|
|
("function-1",),
|
|
tenant_id="tenant-2",
|
|
)
|
|
|
|
def test_batch_accounts_preserve_account_and_subunit_provenance(self) -> None:
|
|
self.directory = SqlIdmDirectory(
|
|
identities=StubIdentityDirectory(
|
|
(
|
|
IdentityRef(
|
|
id="identity-shared",
|
|
primary_account_id="account-1",
|
|
account_ids=("account-1", "account-2"),
|
|
),
|
|
)
|
|
), # type: ignore[arg-type]
|
|
organizations=StubOrganizationDirectory(), # type: ignore[arg-type]
|
|
)
|
|
broad = IdmOrganizationFunctionAssignment(
|
|
id="broad",
|
|
tenant_id="tenant-1",
|
|
identity_id="identity-shared",
|
|
account_id=None,
|
|
function_id="function-1",
|
|
organization_unit_id="unit-1",
|
|
applies_to_subunits=True,
|
|
source="direct",
|
|
is_active=True,
|
|
settings={},
|
|
)
|
|
account_specific = IdmOrganizationFunctionAssignment(
|
|
id="account-specific",
|
|
tenant_id="tenant-1",
|
|
identity_id="identity-shared",
|
|
account_id="account-2",
|
|
function_id="function-2",
|
|
organization_unit_id="unit-1",
|
|
source="governance",
|
|
is_active=True,
|
|
settings={},
|
|
)
|
|
simultaneous = IdmOrganizationFunctionAssignment(
|
|
id="second-incumbent",
|
|
tenant_id="tenant-1",
|
|
identity_id="identity-second",
|
|
function_id="function-1",
|
|
organization_unit_id="unit-1",
|
|
source="direct",
|
|
is_active=True,
|
|
settings={},
|
|
)
|
|
with self.database.session() as session:
|
|
session.add_all((broad, account_specific, simultaneous))
|
|
session.commit()
|
|
|
|
resolved = self.directory.organization_function_assignments_for_accounts(
|
|
("account-1", "account-2", "missing"),
|
|
tenant_id="tenant-1",
|
|
)
|
|
self.assertEqual(("broad",), tuple(item.id for item in resolved["account-1"]))
|
|
self.assertEqual(
|
|
("broad", "account-specific"),
|
|
tuple(item.id for item in resolved["account-2"]),
|
|
)
|
|
self.assertEqual((), resolved["missing"])
|
|
self.assertTrue(resolved["account-1"][0].applies_to_subunits)
|
|
self.assertEqual("governance", resolved["account-2"][1].source)
|
|
|
|
incumbency = self.directory.organization_function_incumbencies(
|
|
("function-1",),
|
|
tenant_id="tenant-1",
|
|
)["function-1"]
|
|
self.assertEqual(
|
|
("broad", "second-incumbent"),
|
|
tuple(item.id for item in incumbency.assignments),
|
|
)
|
|
self.assertFalse(incumbency.vacant)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|