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

Implement typed effective identity relationships

Browse Source
This commit is contained in:
zemion
2026-08-02 14:44:42 +02:00
parent 820ea5eeab
commit 6f6c45f6e2
13 changed files with 1901 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_assignment_expiry.py
+58
View File
@@ -12,7 +12,9 @@ from govoplan_idm.backend.assignment_lifecycle import SqlIdmAssignmentLifecycle
from govoplan_idm.backend.db.models import (
IdmFunctionAssignmentChange,
IdmFunctionAssignmentChangeEvent,
IdmIdentityRelationship,
IdmOrganizationFunctionAssignment,
IdmTypedGroup,
)
from govoplan_organizations.backend.db import models as organization_models # noqa: F401
@@ -26,6 +28,8 @@ class AssignmentExpiryTests(unittest.TestCase):
IdmOrganizationFunctionAssignment.__table__,
IdmFunctionAssignmentChange.__table__,
IdmFunctionAssignmentChangeEvent.__table__,
IdmTypedGroup.__table__,
IdmIdentityRelationship.__table__,
ChangeSequenceEntry.__table__,
],
)
@@ -204,6 +208,60 @@ class AssignmentExpiryTests(unittest.TestCase):
history = session.query(IdmFunctionAssignmentChangeEvent).all()
self.assertEqual(["expired"], [item.action for item in history])
def test_sweep_emits_relationship_expiry_once(self) -> None:
boundary = datetime(2026, 8, 2, 12, tzinfo=timezone.utc)
with self.database.session() as session:
session.add(
IdmTypedGroup(
id="group-1",
tenant_id="tenant-1",
key="eligible",
name="Eligible",
group_type="business_status",
)
)
session.add(
IdmIdentityRelationship(
id="relationship-due",
tenant_id="tenant-1",
relationship_kind="member",
subject_identity_id="identity-1",
target_group_id="group-1",
valid_until=boundary - timedelta(seconds=1),
status="active",
properties={},
provenance={},
)
)
session.commit()
events: list[PlatformEvent] = []
bus = EventBus()
bus.subscribe("idm.relationship.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()
repeated = self.lifecycle.process_expired(
session,
tenant_id="tenant-1",
effective_at=boundary,
)
session.commit()
self.assertEqual(["relationship-due"], result["relationship_ids"])
self.assertEqual(1, result["expired_relationships"])
self.assertEqual(0, repeated["expired_relationships"])
self.assertEqual(1, len(events))
self.assertEqual("identity-1", events[0].subject.id)
with self.database.session() as session:
item = session.get(IdmIdentityRelationship, "relationship-due")
self.assertEqual(2, item.revision)
self.assertIsNotNone(item.expired_event_at)
if __name__ == "__main__":
unittest.main()
tests/test_migrations.py
+58
View File
@@ -0,0 +1,58 @@
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from alembic.runtime.migration import MigrationContext
from sqlalchemy import create_engine, inspect
from govoplan_core.db.migrations import migrate_database
from govoplan_identity.backend.manifest import get_manifest as identity_manifest
from govoplan_idm.backend.manifest import get_manifest as idm_manifest
from govoplan_organizations.backend.manifest import (
get_manifest as organizations_manifest,
)
class IdmMigrationTests(unittest.TestCase):
def test_migrations_create_typed_relationship_tables_and_head(self) -> None:
with tempfile.TemporaryDirectory(prefix="govoplan-idm-migration-") as directory:
url = f"sqlite:///{Path(directory) / 'idm.db'}"
migrate_database(
database_url=url,
enabled_modules=("identity", "organizations", "idm"),
manifest_factories=(
identity_manifest,
organizations_manifest,
idm_manifest,
),
)
engine = create_engine(url)
try:
with engine.connect() as connection:
self.assertIn(
"b1c2d3e4f5a6",
set(MigrationContext.configure(connection).get_current_heads()),
)
self.assertEqual(
{
"idm_function_assignment_change_events",
"idm_function_assignment_changes",
"idm_identity_relationships",
"idm_organization_function_assignments",
"idm_tenant_settings",
"idm_typed_groups",
},
{
name
for name in inspect(connection).get_table_names()
if name.startswith("idm_")
},
)
finally:
engine.dispose()
if __name__ == "__main__":
unittest.main()
tests/test_relationships.py
+220
View File
@@ -0,0 +1,220 @@
from __future__ import annotations
import unittest
from datetime import datetime, timedelta, timezone
from govoplan_core.core.identity import IdentityAccountLinkRef, IdentityRef
from govoplan_core.db.base import Base
from govoplan_core.db.session import configure_database, reset_database
from govoplan_identity.backend.db.models import CanonicalIdentity
from govoplan_idm.backend.db.models import (
IdmIdentityRelationship,
IdmTypedGroup,
)
from govoplan_idm.backend.relationships import SqlIdmRelationshipDirectory
class StubIdentityDirectory:
def __init__(self, identities: tuple[IdentityRef, ...]) -> None:
self._identities = {item.id: item for item in identities}
def get_identity(self, identity_id: str) -> IdentityRef | None:
return self._identities.get(identity_id)
def identity_for_account(self, account_id: str) -> IdentityRef | None:
return None
def identities_for_accounts(self, account_ids):
return ()
def accounts_for_identity(self, identity_id: str) -> tuple[IdentityAccountLinkRef, ...]:
return ()
class IdmRelationshipDirectoryTests(unittest.TestCase):
def setUp(self) -> None:
self.database = configure_database("sqlite:///:memory:")
Base.metadata.create_all(
self.database.engine,
tables=[
CanonicalIdentity.__table__,
IdmTypedGroup.__table__,
IdmIdentityRelationship.__table__,
],
)
identities = (
IdentityRef(id="identity-active", display_name="Active", status="active"),
IdentityRef(id="identity-future", display_name="Future", status="active"),
IdentityRef(id="identity-expired", display_name="Expired", status="active"),
IdentityRef(id="identity-revoked", display_name="Revoked", status="active"),
IdentityRef(id="identity-suspended", display_name="Suspended", status="suspended"),
)
self.directory = SqlIdmRelationshipDirectory(
identities=StubIdentityDirectory(identities) # type: ignore[arg-type]
)
with self.database.session() as session:
session.add_all(
CanonicalIdentity(
id=item.id,
display_name=item.display_name,
source="local",
is_active=item.status == "active",
settings={},
)
for item in identities
)
session.add_all(
(
IdmTypedGroup(
id="group-1",
tenant_id="tenant-1",
key="permit-holder",
name="Permit holders",
group_type="business_status",
source_provider="ldap",
source_resource_type="group",
source_resource_id="cn=permit-holders,dc=example",
source_revision="directory-42",
properties={"classification": "resident"},
provenance={"connector_id": "ldap-1"},
),
IdmTypedGroup(
id="group-other",
tenant_id="tenant-2",
key="other",
name="Other tenant",
group_type="business_status",
),
)
)
session.commit()
def tearDown(self) -> None:
reset_database(dispose=True)
@staticmethod
def _relationship(
relationship_id: str,
identity_id: str,
*,
boundary: datetime,
valid_from: datetime | None = None,
valid_until: datetime | None = None,
status: str = "active",
) -> IdmIdentityRelationship:
return IdmIdentityRelationship(
id=relationship_id,
tenant_id="tenant-1",
relationship_kind="member",
subject_identity_id=identity_id,
target_group_id="group-1",
valid_from=valid_from,
valid_until=valid_until,
status=status,
revoked_at=boundary if status == "revoked" else None,
revoked_by="account-1" if status == "revoked" else None,
revocation_reason="No longer eligible" if status == "revoked" else None,
source_provider="ldap",
source_resource_type="membership",
source_resource_id=f"member:{identity_id}",
source_revision="directory-42",
properties={"rank": 1},
provenance={"sync_run_id": "sync-1"},
revision=1,
)
def test_resolution_explains_current_future_expired_revoked_and_lifecycle(self) -> None:
boundary = datetime(2026, 8, 2, 12, tzinfo=timezone.utc)
with self.database.session() as session:
session.add_all(
(
self._relationship("active", "identity-active", boundary=boundary),
self._relationship(
"future",
"identity-future",
boundary=boundary,
valid_from=boundary + timedelta(days=1),
),
self._relationship(
"expired",
"identity-expired",
boundary=boundary,
valid_until=boundary,
),
self._relationship(
"revoked",
"identity-revoked",
boundary=boundary,
status="revoked",
),
self._relationship(
"suspended",
"identity-suspended",
boundary=boundary,
),
)
)
session.commit()
resolved = self.directory.resolve_typed_group_memberships(
("group-1",),
tenant_id="tenant-1",
effective_at=boundary,
)["group-1"]
self.assertEqual(("identity-active",), resolved.identity_ids)
self.assertEqual(
{
"active": "relationship.effective",
"future": "relationship.not_yet_effective",
"expired": "relationship.expired",
"revoked": "relationship.revoked",
"suspended": "identity.not_active",
},
{item.relationship.id: item.code for item in resolved.decisions},
)
self.assertEqual("directory-42", resolved.group.source_revision)
self.assertEqual("ldap", resolved.decisions[0].relationship.source_provider)
def test_forward_reverse_batch_queries_return_only_effective_relationships(self) -> None:
boundary = datetime(2026, 8, 2, 12, tzinfo=timezone.utc)
with self.database.session() as session:
session.add_all(
(
self._relationship("active", "identity-active", boundary=boundary),
self._relationship(
"future",
"identity-future",
boundary=boundary,
valid_from=boundary + timedelta(days=1),
),
)
)
session.commit()
forward = self.directory.identity_relationships_for_identities(
("identity-active", "identity-future"),
tenant_id="tenant-1",
effective_at=boundary,
)
reverse = self.directory.identity_relationships_for_groups(
("group-1",),
tenant_id="tenant-1",
effective_at=boundary,
)
self.assertEqual(("active",), tuple(item.id for item in forward["identity-active"]))
self.assertEqual((), forward["identity-future"])
self.assertEqual(("active",), tuple(item.id for item in reverse["group-1"]))
def test_cross_tenant_group_references_are_rejected(self) -> None:
with self.assertRaisesRegex(ValueError, "another tenant"):
self.directory.get_typed_group("group-other", tenant_id="tenant-1")
with self.assertRaisesRegex(ValueError, "another tenant"):
self.directory.identity_relationships_for_group(
"group-other", tenant_id="tenant-1"
)
if __name__ == "__main__":
unittest.main()