221 lines
8.4 KiB
Python
221 lines
8.4 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.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()
|