139 lines
5.3 KiB
Python
139 lines
5.3 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 get_identity(self, identity_id: str) -> IdentityRef | None:
|
|
return None
|
|
|
|
def identity_for_account(self, account_id: str) -> IdentityRef | None:
|
|
return None
|
|
|
|
def identities_for_accounts(self, account_ids: tuple[str, ...]) -> tuple[IdentityRef, ...]:
|
|
return ()
|
|
|
|
def accounts_for_identity(self, identity_id: str) -> tuple[IdentityAccountLinkRef, ...]:
|
|
return ()
|
|
|
|
|
|
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))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|