refactor(idm): consume identity and organization contracts

tests/test_directory.py

@@ -0,0 +1,138 @@
+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()