GovOPlaN/govoplan-idm
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

Complete effective assignment expiry lifecycle

Browse Source
This commit is contained in:
zemion
2026-07-31 18:07:36 +02:00
parent d1c5738ca8
commit f025b0c25b
13 changed files with 619 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+6
View File
@@ -133,6 +133,12 @@ change record before the effective assignment is created. The target journeys,
grant profiles, state model, and module boundaries are documented in
[Function assignment request and grant workflows](docs/FUNCTION_ASSIGNMENT_WORKFLOWS.md).
Assignments with a future `valid_until` are resolved as ineffective immediately
after that boundary. When Celery beat and an IDM-capable worker are running, the
shared `govoplan.idm.expire_assignments` task also emits the corresponding
`idm.function_assignment.expired.v1` event. IDM records the event marker in the
same database transaction, making repeated sweeps idempotent.
## First Milestone
The first useful milestone is a read-only synchronization preview:
src/govoplan_idm/backend/api/v1/routes.py
+27 -50
View File
@@ -16,13 +16,6 @@ from govoplan_core.core.configuration_control import (
ensure_configuration_change_allowed,
record_configuration_change_applied,
)
from govoplan_core.core.events import (
EventActorRef,
EventObjectRef,
EventTenantRef,
PlatformEvent,
emit_platform_event,
)
from govoplan_core.core.principal_cache import invalidate_auth_principals
from govoplan_core.core.identity import (
CAPABILITY_IDENTITY_DIRECTORY,
@@ -42,11 +35,13 @@ from govoplan_core.security.time import utc_now
from govoplan_idm.backend.assignment_transitions import (
AssignmentMutationPlan,
AssignmentTransitionError,
assignment_is_expired,
lifecycle_event_types,
plan_assignment_update,
validate_assignment_shape,
validate_assignment_source_rules,
)
from govoplan_idm.backend.assignment_events import emit_assignment_event
from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment, IdmTenantSettings
from .schemas import (
@@ -436,45 +431,12 @@ def _publish_assignment_event(
*,
event_type: str,
) -> None:
emit_platform_event(
emit_assignment_event(
session,
PlatformEvent(
type=event_type,
module_id="idm",
payload={
"identity_id": item.identity_id,
"account_id": item.account_id,
"function_id": item.function_id,
"organization_unit_id": item.organization_unit_id,
"source": item.source,
"delegated_from_assignment_id": (
item.delegated_from_assignment_id
),
"acting_for_account_id": item.acting_for_account_id,
"valid_from": (
item.valid_from.isoformat()
if item.valid_from is not None
else None
),
"valid_until": (
item.valid_until.isoformat()
if item.valid_until is not None
else None
),
"is_active": item.is_active,
},
actor=EventActorRef(type="account", id=principal.account_id),
tenant=EventTenantRef(id=principal.tenant_id),
subject=EventObjectRef(
type="organization_function",
id=item.function_id,
),
resource=EventObjectRef(
type="organization_function_assignment",
id=item.id,
),
classification="internal",
)
item,
event_type=event_type,
actor_type="account",
actor_id=principal.account_id,
)
@@ -621,6 +583,15 @@ def create_organization_function_assignment(
result,
event_type="idm.function_assignment.created.v1",
)
now = utc_now()
if assignment_is_expired(item, now=now):
item.expired_event_at = now
_publish_assignment_event(
session,
principal,
_assignment_item(item),
event_type="idm.function_assignment.expired.v1",
)
_commit_assignment_transaction(session, item)
return _assignment_item(item)
@@ -643,6 +614,16 @@ def update_organization_function_assignment(
tenant_id=tenant_id,
)
plan.apply(item)
now = utc_now()
event_types = lifecycle_event_types(
plan.before,
plan.after,
now=now,
)
if "idm.function_assignment.expired.v1" in event_types:
item.expired_event_at = now
elif not assignment_is_expired(item, now=now):
item.expired_event_at = None
_flush_assignment(session, item)
result = _assignment_item(item)
after = result.model_dump(mode="json")
@@ -656,11 +637,7 @@ def update_organization_function_assignment(
before=before,
after=after,
)
for event_type in lifecycle_event_types(
plan.before,
plan.after,
now=utc_now(),
):
for event_type in event_types:
_publish_assignment_event(
session,
principal,
src/govoplan_idm/backend/api/v1/schemas.py
+1
View File
@@ -37,6 +37,7 @@ class OrganizationFunctionAssignmentItem(BaseModel):
acting_for_account_id: str | None = None
valid_from: datetime | None = None
valid_until: datetime | None = None
expired_event_at: datetime | None = None
is_active: bool
settings: dict[str, Any]
created_at: datetime
src/govoplan_idm/backend/assignment_events.py
+87
View File
@@ -0,0 +1,87 @@
from __future__ import annotations
from datetime import datetime
from typing import Protocol
from sqlalchemy.orm import Session
from govoplan_core.core.events import (
EventActorRef,
EventObjectRef,
EventTenantRef,
PlatformEvent,
emit_platform_event,
)
class AssignmentEventSource(Protocol):
id: str
tenant_id: str
identity_id: str
account_id: str | None
function_id: str
organization_unit_id: str
applies_to_subunits: bool
source: str
delegated_from_assignment_id: str | None
acting_for_account_id: str | None
valid_from: datetime | None
valid_until: datetime | None
is_active: bool
def emit_assignment_event(
session: Session,
item: AssignmentEventSource,
*,
event_type: str,
actor_type: str,
actor_id: str | None = None,
occurred_at: datetime | None = None,
) -> None:
event_options = {"occurred_at": occurred_at} if occurred_at else {}
emit_platform_event(
session,
PlatformEvent(
type=event_type,
module_id="idm",
payload={
"identity_id": item.identity_id,
"account_id": item.account_id,
"function_id": item.function_id,
"organization_unit_id": item.organization_unit_id,
"applies_to_subunits": item.applies_to_subunits,
"source": item.source,
"delegated_from_assignment_id": (
item.delegated_from_assignment_id
),
"acting_for_account_id": item.acting_for_account_id,
"valid_from": (
item.valid_from.isoformat()
if item.valid_from is not None
else None
),
"valid_until": (
item.valid_until.isoformat()
if item.valid_until is not None
else None
),
"is_active": item.is_active,
},
actor=EventActorRef(type=actor_type, id=actor_id),
tenant=EventTenantRef(id=item.tenant_id),
subject=EventObjectRef(
type="organization_function",
id=item.function_id,
),
resource=EventObjectRef(
type="organization_function_assignment",
id=item.id,
),
classification="internal",
**event_options,
),
)
__all__ = ["emit_assignment_event"]
src/govoplan_idm/backend/assignment_lifecycle.py
+93
View File
@@ -0,0 +1,93 @@
from __future__ import annotations
from datetime import datetime
from sqlalchemy.orm import Session
from govoplan_core.core.principal_cache import invalidate_auth_principals
from govoplan_core.security.time import ensure_aware_utc, utc_now
from govoplan_idm.backend.assignment_events import emit_assignment_event
from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment
class SqlIdmAssignmentLifecycle:
"""Claim and publish elapsed assignments exactly once per validity window."""
def process_expired(
self,
session: object,
*,
tenant_id: str | None = None,
effective_at: datetime | None = None,
limit: int = 100,
) -> dict[str, object]:
if not isinstance(session, Session):
raise TypeError("IDM assignment lifecycle requires a SQLAlchemy session")
if limit < 1 or limit > 1000:
raise ValueError("IDM assignment expiry limit must be between 1 and 1000")
now = ensure_aware_utc(effective_at) or utc_now()
query = session.query(IdmOrganizationFunctionAssignment).filter(
IdmOrganizationFunctionAssignment.is_active.is_(True),
IdmOrganizationFunctionAssignment.valid_until.is_not(None),
IdmOrganizationFunctionAssignment.valid_until <= now,
IdmOrganizationFunctionAssignment.expired_event_at.is_(None),
)
if tenant_id is not None:
query = query.filter(
IdmOrganizationFunctionAssignment.tenant_id == tenant_id
)
candidates = (
query.order_by(
IdmOrganizationFunctionAssignment.valid_until.asc(),
IdmOrganizationFunctionAssignment.id.asc(),
)
.limit(limit)
.all()
)
expired_ids: list[str] = []
touched_tenants: set[str] = set()
for item in candidates:
claimed = (
session.query(IdmOrganizationFunctionAssignment)
.filter(
IdmOrganizationFunctionAssignment.id == item.id,
IdmOrganizationFunctionAssignment.is_active.is_(True),
IdmOrganizationFunctionAssignment.valid_until.is_not(None),
IdmOrganizationFunctionAssignment.valid_until <= now,
IdmOrganizationFunctionAssignment.expired_event_at.is_(None),
)
.update(
{IdmOrganizationFunctionAssignment.expired_event_at: now},
synchronize_session=False,
)
)
if claimed != 1:
continue
session.refresh(item)
emit_assignment_event(
session,
item,
event_type="idm.function_assignment.expired.v1",
actor_type="system",
occurred_at=now,
)
expired_ids.append(item.id)
touched_tenants.add(item.tenant_id)
for touched_tenant_id in sorted(touched_tenants):
invalidate_auth_principals(
session,
tenant_id=touched_tenant_id,
source_module="idm",
resource_type="organization_function_assignment_expiry",
resource_id=touched_tenant_id,
)
return {
"selected": len(candidates),
"expired": len(expired_ids),
"assignment_ids": expired_ids,
}
__all__ = ["SqlIdmAssignmentLifecycle"]
src/govoplan_idm/backend/assignment_transitions.py
+25 -4
View File
@@ -200,15 +200,35 @@ def lifecycle_event_types(
events = ["idm.function_assignment.changed.v1"]
if before.is_active and not after.is_active:
events.append("idm.function_assignment.revoked.v1")
if (
after.valid_until is not None
and _comparable_datetime(after.valid_until) <= _comparable_datetime(now)
and before.valid_until != after.valid_until
before_expired = _is_expired(before, now=now)
after_expired = _is_expired(after, now=now)
if after_expired and (
not before_expired or before.valid_until != after.valid_until
):
events.append("idm.function_assignment.expired.v1")
return tuple(events)
def assignment_is_expired(
item: AssignmentLike | AssignmentSnapshot,
*,
now: datetime,
) -> bool:
return _is_expired(item, now=now)
def _is_expired(
item: AssignmentLike | AssignmentSnapshot,
*,
now: datetime,
) -> bool:
return bool(
item.is_active
and item.valid_until is not None
and _comparable_datetime(item.valid_until) <= _comparable_datetime(now)
)
def _validate_delegated_assignment(
item: AssignmentLike,
*,
@@ -288,6 +308,7 @@ __all__ = [
"AssignmentMutationPlan",
"AssignmentSnapshot",
"AssignmentTransitionError",
"assignment_is_expired",
"lifecycle_event_types",
"plan_assignment_update",
"validate_assignment_shape",
src/govoplan_idm/backend/db/models.py
+8 -1
View File
@@ -4,7 +4,7 @@ import uuid
from datetime import datetime
from typing import Any
from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, JSON, String, UniqueConstraint
from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, JSON, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from govoplan_core.db.base import Base, TimestampMixin
@@ -24,6 +24,12 @@ class IdmOrganizationFunctionAssignment(Base, TimestampMixin):
"organization_unit_id",
name="uq_idm_org_function_assignments_identity_scope",
),
Index(
"ix_idm_org_function_assignments_expiry_due",
"is_active",
"expired_event_at",
"valid_until",
),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
@@ -38,6 +44,7 @@ class IdmOrganizationFunctionAssignment(Base, TimestampMixin):
acting_for_account_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
valid_from: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
valid_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
expired_event_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
settings: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)
src/govoplan_idm/backend/manifest.py
+16
View File
@@ -5,6 +5,7 @@ from pathlib import Path
from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER
from govoplan_core.core.identity import CAPABILITY_IDENTITY_DIRECTORY, CAPABILITY_IDENTITY_SEARCH, IdentityDirectory
from govoplan_core.core.idm import (
CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE,
CAPABILITY_IDM_DIRECTORY,
CAPABILITY_IDM_FUNCTION_ASSIGNMENTS,
)
@@ -111,6 +112,15 @@ def _idm_directory(context: ModuleContext) -> object:
return SqlIdmDirectory(identities=identities, organizations=organizations)
def _assignment_lifecycle(context: ModuleContext) -> object:
del context
from govoplan_idm.backend.assignment_lifecycle import (
SqlIdmAssignmentLifecycle,
)
return SqlIdmAssignmentLifecycle()
manifest = ModuleManifest(
id="idm",
name="IDM",
@@ -129,6 +139,10 @@ manifest = ModuleManifest(
name=CAPABILITY_IDM_FUNCTION_ASSIGNMENTS,
version="0.1.8",
),
ModuleInterfaceProvider(
name=CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE,
version="0.1.8",
),
),
permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES,
@@ -162,6 +176,7 @@ manifest = ModuleManifest(
),
),
capability_factories={
CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE: _assignment_lifecycle,
CAPABILITY_IDM_DIRECTORY: _idm_directory,
CAPABILITY_IDM_FUNCTION_ASSIGNMENTS: _idm_directory,
},
@@ -188,6 +203,7 @@ manifest = ModuleManifest(
body=(
"Assignment links are high-impact because they can later feed access decisions. "
"Tenants can enable recorded change requests for assignment create and update operations. "
"A periodic worker emits one expiry event when a future-dated assignment elapses; the marker and event are committed together so retries remain idempotent. "
"The legacy organizations:function:assign scope remains accepted for transition, while new role templates should grant idm:organization_assignment:write."
),
layer="configured",
src/govoplan_idm/backend/migrations/dev_versions/9a0b1c2d3e4f_idm_assignment_expiry_events.py
+40
View File
@@ -0,0 +1,40 @@
"""Track emitted assignment expiry events.
Revision ID: 9a0b1c2d3e4f
Revises: 8f9a0b1c2d3e
Create Date: 2026-07-31 00:00:00.000000
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "9a0b1c2d3e4f"
down_revision = "8f9a0b1c2d3e"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.add_column(
"idm_organization_function_assignments",
sa.Column("expired_event_at", sa.DateTime(timezone=True), nullable=True),
)
op.create_index(
"ix_idm_org_function_assignments_expiry_due",
"idm_organization_function_assignments",
["is_active", "expired_event_at", "valid_until"],
unique=False,
)
def downgrade() -> None:
op.drop_index(
"ix_idm_org_function_assignments_expiry_due",
table_name="idm_organization_function_assignments",
)
op.drop_column(
"idm_organization_function_assignments",
"expired_event_at",
)
src/govoplan_idm/backend/migrations/versions/9a0b1c2d3e4f_idm_assignment_expiry_events.py
+40
View File
@@ -0,0 +1,40 @@
"""Track emitted assignment expiry events.
Revision ID: 9a0b1c2d3e4f
Revises: 8f9a0b1c2d3e
Create Date: 2026-07-31 00:00:00.000000
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "9a0b1c2d3e4f"
down_revision = "8f9a0b1c2d3e"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.add_column(
"idm_organization_function_assignments",
sa.Column("expired_event_at", sa.DateTime(timezone=True), nullable=True),
)
op.create_index(
"ix_idm_org_function_assignments_expiry_due",
"idm_organization_function_assignments",
["is_active", "expired_event_at", "valid_until"],
unique=False,
)
def downgrade() -> None:
op.drop_index(
"ix_idm_org_function_assignments_expiry_due",
table_name="idm_organization_function_assignments",
)
op.drop_column(
"idm_organization_function_assignments",
"expired_event_at",
)
tests/test_assignment_expiry.py
+148
View File
@@ -0,0 +1,148 @@
from __future__ import annotations
import unittest
from datetime import datetime, timedelta, timezone
from govoplan_core.core.change_sequence import ChangeSequenceEntry
from govoplan_core.core.events import EventBus, PlatformEvent, event_bus_context
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
from govoplan_idm.backend.assignment_lifecycle import SqlIdmAssignmentLifecycle
from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment
from govoplan_organizations.backend.db import models as organization_models # noqa: F401
class AssignmentExpiryTests(unittest.TestCase):
def setUp(self) -> None:
self.database = configure_database("sqlite:///:memory:")
Base.metadata.create_all(
self.database.engine,
tables=[
IdmOrganizationFunctionAssignment.__table__,
ChangeSequenceEntry.__table__,
],
)
self.lifecycle = SqlIdmAssignmentLifecycle()
def tearDown(self) -> None:
reset_database(dispose=True)
@staticmethod
def _assignment(
assignment_id: str,
*,
tenant_id: str = "tenant-1",
valid_until: datetime,
active: bool = True,
expired_event_at: datetime | None = None,
) -> IdmOrganizationFunctionAssignment:
return IdmOrganizationFunctionAssignment(
id=assignment_id,
tenant_id=tenant_id,
identity_id=f"identity-{assignment_id}",
function_id="function-1",
organization_unit_id="unit-1",
source="direct",
valid_until=valid_until,
expired_event_at=expired_event_at,
is_active=active,
settings={},
)
def test_sweep_claims_due_assignments_once_and_preserves_provenance(self) -> None:
boundary = datetime(2026, 7, 31, 12, tzinfo=timezone.utc)
with self.database.session() as session:
session.add_all(
(
self._assignment(
"due",
valid_until=boundary - timedelta(seconds=1),
),
self._assignment(
"future",
valid_until=boundary + timedelta(seconds=1),
),
self._assignment(
"revoked",
valid_until=boundary - timedelta(seconds=1),
active=False,
),
self._assignment(
"already-emitted",
valid_until=boundary - timedelta(seconds=1),
expired_event_at=boundary - timedelta(minutes=1),
),
self._assignment(
"other-tenant",
tenant_id="tenant-2",
valid_until=boundary - timedelta(seconds=1),
),
)
)
session.commit()
events: list[PlatformEvent] = []
bus = EventBus()
bus.subscribe("idm.function_assignment.expired.v1", events.append)
with self.database.SessionLocal() as session, event_bus_context(bus):
result = self.lifecycle.process_expired(
session,
tenant_id="tenant-1",
effective_at=boundary,
)
session.commit()
self.assertEqual(1, result["selected"])
self.assertEqual(1, result["expired"])
self.assertEqual(["due"], result["assignment_ids"])
self.assertEqual(1, len(events))
self.assertEqual("system", events[0].actor.type)
self.assertEqual("tenant-1", events[0].tenant.id)
self.assertEqual("identity-due", events[0].payload["identity_id"])
self.assertEqual("function-1", events[0].payload["function_id"])
with self.database.session() as session:
due = session.get(IdmOrganizationFunctionAssignment, "due")
self.assertEqual(boundary, due.expired_event_at.replace(tzinfo=timezone.utc))
repeated = self.lifecycle.process_expired(
session,
tenant_id="tenant-1",
effective_at=boundary,
)
session.commit()
self.assertEqual(0, repeated["expired"])
self.assertEqual(1, len(events))
def test_sweep_rollback_releases_marker_and_event(self) -> None:
boundary = datetime(2026, 7, 31, 12, tzinfo=timezone.utc)
with self.database.session() as session:
session.add(
self._assignment(
"rolled-back",
valid_until=boundary - timedelta(seconds=1),
)
)
session.commit()
events: list[PlatformEvent] = []
bus = EventBus()
bus.subscribe("idm.function_assignment.expired.v1", events.append)
with self.database.SessionLocal() as session, event_bus_context(bus):
self.lifecycle.process_expired(session, effective_at=boundary)
session.rollback()
self.assertEqual([], events)
with self.database.session() as session:
item = session.get(IdmOrganizationFunctionAssignment, "rolled-back")
self.assertIsNone(item.expired_event_at)
def test_limit_validation_is_bounded(self) -> None:
with self.database.session() as session:
for value in (0, 1001):
with self.subTest(limit=value), self.assertRaises(ValueError):
self.lifecycle.process_expired(session, limit=value)
if __name__ == "__main__":
unittest.main()
tests/test_assignment_workflow.py
+20
View File
@@ -233,6 +233,26 @@ class AssignmentWorkflowTests(unittest.TestCase):
lifecycle_event_types(before, after, now=now),
)
def test_reactivation_of_elapsed_assignment_emits_expiry(self) -> None:
now = datetime.now(timezone.utc)
item = assignment(
is_active=False,
valid_until=now - timedelta(minutes=1),
)
before = AssignmentSnapshot.from_assignment(item) # type: ignore[arg-type]
after = plan_assignment_update( # type: ignore[arg-type]
item,
{"is_active": True},
).after
self.assertEqual(
(
"idm.function_assignment.changed.v1",
"idm.function_assignment.expired.v1",
),
lifecycle_event_types(before, after, now=now),
)
if __name__ == "__main__":
unittest.main()
tests/test_directory.py
+107 -3
View File
@@ -14,17 +14,48 @@ from govoplan_organizations.backend.db import models as organization_models # n
class StubIdentityDirectory:
def __init__(self, identities: tuple[IdentityRef, ...] = ()) -> None:
self.identities = identities
def get_identity(self, identity_id: str) -> IdentityRef | None:
return 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 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, ...]:
return ()
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:
@@ -227,6 +258,79 @@ class IdmDirectoryDelegationTests(unittest.TestCase):
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()