intermittent commit

README.md

@@ -1,5 +1,9 @@
 # GovOPlaN IDM
 
+<!-- govoplan-repository-type:start -->
+**Repository type:** module (platform).
+<!-- govoplan-repository-type:end -->
+
 `govoplan-idm` is the planned integration module for external identity
 management systems. It does not own GovOPlaN's internal identity, organization,
 account, role, or tenant tables; those remain with `govoplan-identity`,

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/idm-webui",
-  "version": "0.1.7",
+  "version": "0.1.8",
   "private": true,
   "type": "module",
   "main": "webui/src/index.ts",
@@ -19,7 +19,7 @@
     "LICENSE"
   ],
   "peerDependencies": {
-    "@govoplan/core-webui": "^0.1.7",
+    "@govoplan/core-webui": "^0.1.8",
     "@vitejs/plugin-react": "^4.3.4",
     "lucide-react": "^1.23.0",
     "react": "^19.0.0",

pyproject.toml

@@ -4,15 +4,15 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-idm"
-version = "0.1.7"
+version = "0.1.8"
 description = "GovOPlaN identity management bridge module."
 readme = "README.md"
 requires-python = ">=3.12"
 authors = [{ name = "GovOPlaN" }]
 dependencies = [
-  "govoplan-core>=0.1.7",
-  "govoplan-identity>=0.1.7",
-  "govoplan-organizations>=0.1.7",
+  "govoplan-core>=0.1.8",
+  "govoplan-identity>=0.1.8",
+  "govoplan-organizations>=0.1.8",
 ]
 
 [tool.setuptools.packages.find]

src/govoplan_idm/backend/api/v1/routes.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Callable
 from typing import Any, TypeVar
 
 from fastapi import APIRouter, Depends, HTTPException, Query, status
@@ -158,6 +159,18 @@ def _ensure_assignment_workflow(
     *,
     tenant_id: str,
 ) -> None:
+    _ensure_assignment_shape(item)
+    function = _get_tenant_row(session, OrganizationFunction, item.function_id, tenant_id, "Organization function")
+    base = _assignment_source_assignment(session, item, tenant_id=tenant_id)
+    _ensure_assignment_source_rules(
+        item,
+        function=function,
+        base=base,
+        account_linked_to_identity=lambda identity_id, account_id: _account_linked_to_identity(session, identity_id, account_id),
+    )
+
+
+def _ensure_assignment_shape(item: IdmOrganizationFunctionAssignment) -> None:
     if item.source not in ASSIGNMENT_SOURCES:
         raise _invalid("Assignment source is not supported.")
     if item.valid_from is not None and item.valid_until is not None and item.valid_until <= item.valid_from:
@@ -165,43 +178,81 @@ def _ensure_assignment_workflow(
     if item.delegated_from_assignment_id is not None and item.delegated_from_assignment_id == item.id:
         raise _invalid("A function assignment cannot delegate from itself.")
 
-    function = _get_tenant_row(session, OrganizationFunction, item.function_id, tenant_id, "Organization function")
-    base: IdmOrganizationFunctionAssignment | None = None
-    if item.delegated_from_assignment_id is not None:
-        base = _get_tenant_row(session, IdmOrganizationFunctionAssignment, item.delegated_from_assignment_id, tenant_id, "Delegated function assignment")
-        if not base.is_active:
-            raise _invalid("Delegation source assignment must be active.")
-        if base.function_id != item.function_id:
-            raise _invalid("Delegated and acting-for assignments must use the same organization function as the source assignment.")
 
+def _assignment_source_assignment(
+    session: Session,
+    item: IdmOrganizationFunctionAssignment,
+    *,
+    tenant_id: str,
+) -> IdmOrganizationFunctionAssignment | None:
+    if item.delegated_from_assignment_id is None:
+        return None
+    base = _get_tenant_row(session, IdmOrganizationFunctionAssignment, item.delegated_from_assignment_id, tenant_id, "Delegated function assignment")
+    if not base.is_active:
+        raise _invalid("Delegation source assignment must be active.")
+    if base.function_id != item.function_id:
+        raise _invalid("Delegated and acting-for assignments must use the same organization function as the source assignment.")
+    return base
+
+
+def _ensure_assignment_source_rules(
+    item: IdmOrganizationFunctionAssignment,
+    *,
+    function: OrganizationFunction,
+    base: IdmOrganizationFunctionAssignment | None,
+    account_linked_to_identity: Callable[[str, str], bool],
+) -> None:
     if item.source == "delegated":
-        if base is None:
-            raise _invalid("Delegated assignments require a source assignment.")
-        if not function.delegable:
-            raise _invalid("This organization function does not allow delegation.")
-        if item.acting_for_account_id is not None:
-            raise _invalid("Delegated assignments cannot set an acting-for account.")
-        if item.identity_id == base.identity_id and (item.account_id or "") == (base.account_id or ""):
-            raise _invalid("A delegated assignment must target another identity or account.")
-    elif item.source == "acting_for":
-        if base is None:
-            raise _invalid("Acting-for assignments require a source assignment.")
-        if not function.act_in_place_allowed:
-            raise _invalid("This organization function does not allow acting in place.")
-        if item.acting_for_account_id is None:
-            raise _invalid("Acting-for assignments require an acting-for account.")
-        if base.account_id is not None:
-            if item.acting_for_account_id != base.account_id:
-                raise _invalid("Acting-for account must match the source assignment account.")
-        elif not _account_linked_to_identity(session, base.identity_id, item.acting_for_account_id):
-            raise _invalid("Acting-for account must belong to the source assignment identity.")
-        if item.account_id == item.acting_for_account_id:
-            raise _invalid("The acting account and acting-for account must be different.")
-    else:
-        if item.delegated_from_assignment_id is not None:
-            raise _invalid("Only delegated or acting-for assignments can reference a source assignment.")
-        if item.acting_for_account_id is not None:
-            raise _invalid("Only acting-for assignments can set an acting-for account.")
+        _ensure_delegated_assignment(item, function=function, base=base)
+        return
+    if item.source == "acting_for":
+        _ensure_acting_for_assignment(item, function=function, base=base, account_linked_to_identity=account_linked_to_identity)
+        return
+    _ensure_direct_assignment_shape(item)
+
+
+def _ensure_delegated_assignment(
+    item: IdmOrganizationFunctionAssignment,
+    *,
+    function: OrganizationFunction,
+    base: IdmOrganizationFunctionAssignment | None,
+) -> None:
+    if base is None:
+        raise _invalid("Delegated assignments require a source assignment.")
+    if not function.delegable:
+        raise _invalid("This organization function does not allow delegation.")
+    if item.acting_for_account_id is not None:
+        raise _invalid("Delegated assignments cannot set an acting-for account.")
+    if item.identity_id == base.identity_id and (item.account_id or "") == (base.account_id or ""):
+        raise _invalid("A delegated assignment must target another identity or account.")
+
+
+def _ensure_acting_for_assignment(
+    item: IdmOrganizationFunctionAssignment,
+    *,
+    function: OrganizationFunction,
+    base: IdmOrganizationFunctionAssignment | None,
+    account_linked_to_identity: Callable[[str, str], bool],
+) -> None:
+    if base is None:
+        raise _invalid("Acting-for assignments require a source assignment.")
+    if not function.act_in_place_allowed:
+        raise _invalid("This organization function does not allow acting in place.")
+    if item.acting_for_account_id is None:
+        raise _invalid("Acting-for assignments require an acting-for account.")
+    if base.account_id is not None and item.acting_for_account_id != base.account_id:
+        raise _invalid("Acting-for account must match the source assignment account.")
+    if base.account_id is None and not account_linked_to_identity(base.identity_id, item.acting_for_account_id):
+        raise _invalid("Acting-for account must belong to the source assignment identity.")
+    if item.account_id == item.acting_for_account_id:
+        raise _invalid("The acting account and acting-for account must be different.")
+
+
+def _ensure_direct_assignment_shape(item: IdmOrganizationFunctionAssignment) -> None:
+    if item.delegated_from_assignment_id is not None:
+        raise _invalid("Only delegated or acting-for assignments can reference a source assignment.")
+    if item.acting_for_account_id is not None:
+        raise _invalid("Only acting-for assignments can set an acting-for account.")
 
 
 def _requires_assignment_change_request(session: Session, tenant_id: str) -> bool:

src/govoplan_idm/backend/manifest.py

@@ -102,7 +102,7 @@ def _idm_directory(context: ModuleContext) -> object:
 manifest = ModuleManifest(
     id="idm",
     name="IDM",
-    version="0.1.7",
+    version="0.1.8",
     dependencies=("identity", "organizations"),
     optional_dependencies=("access", "audit"),
     required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),

src/govoplan_idm/backend/migrations/dev_versions/__init__.py

@@ -0,0 +1 @@
+"""IDM migration revisions."""

src/govoplan_idm/backend/migrations/versions/8f9a0b1c2d3e_v017_idm_baseline.py

@@ -0,0 +1,65 @@
+"""v0.1.7 idm baseline
+
+Revision ID: 8f9a0b1c2d3e
+Revises: None
+Create Date: 2026-07-11 00:00:00.000000
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = '8f9a0b1c2d3e'
+down_revision = None
+branch_labels = None
+depends_on = ('5c6d7e8f9a10', '6d7e8f9a0b1c')
+
+
+def upgrade() -> None:
+    op.create_table('idm_tenant_settings',
+    sa.Column('tenant_id', sa.String(length=36), nullable=False),
+    sa.Column('require_assignment_change_requests', sa.Boolean(), nullable=False),
+    sa.Column('audit_detail_level', sa.String(length=20), nullable=False),
+    sa.Column('change_retention_days', sa.Integer(), nullable=True),
+    sa.Column('settings', sa.JSON(), nullable=False),
+    sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+    sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False),
+    sa.PrimaryKeyConstraint('tenant_id', name=op.f('pk_idm_tenant_settings'))
+    )
+    op.create_table('idm_organization_function_assignments',
+    sa.Column('id', sa.String(length=36), nullable=False),
+    sa.Column('tenant_id', sa.String(length=36), nullable=False),
+    sa.Column('identity_id', sa.String(length=36), nullable=False),
+    sa.Column('account_id', sa.String(length=36), nullable=True),
+    sa.Column('function_id', sa.String(length=36), nullable=False),
+    sa.Column('organization_unit_id', sa.String(length=36), nullable=False),
+    sa.Column('applies_to_subunits', sa.Boolean(), nullable=False),
+    sa.Column('source', sa.String(length=50), nullable=False),
+    sa.Column('delegated_from_assignment_id', sa.String(length=36), nullable=True),
+    sa.Column('acting_for_account_id', sa.String(length=36), nullable=True),
+    sa.Column('valid_from', sa.DateTime(timezone=True), nullable=True),
+    sa.Column('valid_until', sa.DateTime(timezone=True), nullable=True),
+    sa.Column('is_active', sa.Boolean(), nullable=False),
+    sa.Column('settings', sa.JSON(), nullable=False),
+    sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+    sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False),
+    sa.ForeignKeyConstraint(['delegated_from_assignment_id'], ['idm_organization_function_assignments.id'], name=op.f('fk_idm_organization_function_assignments_delegated_from_assignment_id_idm_organization_function_assignments'), ondelete='SET NULL'),
+    sa.ForeignKeyConstraint(['function_id'], ['organizations_functions.id'], name=op.f('fk_idm_organization_function_assignments_function_id_organizations_functions'), ondelete='CASCADE'),
+    sa.ForeignKeyConstraint(['identity_id'], ['identity_identities.id'], name=op.f('fk_idm_organization_function_assignments_identity_id_identity_identities'), ondelete='CASCADE'),
+    sa.ForeignKeyConstraint(['organization_unit_id'], ['organizations_units.id'], name=op.f('fk_idm_organization_function_assignments_organization_unit_id_organizations_units'), ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id', name=op.f('pk_idm_organization_function_assignments')),
+    sa.UniqueConstraint('tenant_id', 'identity_id', 'function_id', 'organization_unit_id', name='uq_idm_org_function_assignments_identity_scope')
+    )
+    op.create_index(op.f('ix_idm_organization_function_assignments_account_id'), 'idm_organization_function_assignments', ['account_id'], unique=False)
+    op.create_index(op.f('ix_idm_organization_function_assignments_acting_for_account_id'), 'idm_organization_function_assignments', ['acting_for_account_id'], unique=False)
+    op.create_index(op.f('ix_idm_organization_function_assignments_delegated_from_assignment_id'), 'idm_organization_function_assignments', ['delegated_from_assignment_id'], unique=False)
+    op.create_index(op.f('ix_idm_organization_function_assignments_function_id'), 'idm_organization_function_assignments', ['function_id'], unique=False)
+    op.create_index(op.f('ix_idm_organization_function_assignments_identity_id'), 'idm_organization_function_assignments', ['identity_id'], unique=False)
+    op.create_index(op.f('ix_idm_organization_function_assignments_organization_unit_id'), 'idm_organization_function_assignments', ['organization_unit_id'], unique=False)
+    op.create_index(op.f('ix_idm_organization_function_assignments_tenant_id'), 'idm_organization_function_assignments', ['tenant_id'], unique=False)
+
+
+def downgrade() -> None:
+    op.drop_table('idm_organization_function_assignments')
+    op.drop_table('idm_tenant_settings')

tests/test_assignment_workflow.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+import unittest
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+
+from fastapi import HTTPException
+
+from govoplan_idm.backend.api.v1.routes import _ensure_assignment_shape, _ensure_assignment_source_rules
+
+
+def assignment(**overrides: object) -> SimpleNamespace:
+    values = {
+        "id": "assignment-1",
+        "identity_id": "identity-1",
+        "account_id": "account-1",
+        "function_id": "function-1",
+        "source": "direct",
+        "delegated_from_assignment_id": None,
+        "acting_for_account_id": None,
+        "valid_from": None,
+        "valid_until": None,
+    }
+    values.update(overrides)
+    return SimpleNamespace(**values)
+
+
+def function(**overrides: object) -> SimpleNamespace:
+    values = {"delegable": True, "act_in_place_allowed": True}
+    values.update(overrides)
+    return SimpleNamespace(**values)
+
+
+class AssignmentWorkflowTests(unittest.TestCase):
+    def assert_invalid(self, expected_detail: str, callback: object) -> None:
+        with self.assertRaises(HTTPException) as captured:
+            callback()
+        self.assertEqual(captured.exception.detail, expected_detail)
+
+    def test_direct_assignment_accepts_plain_shape(self) -> None:
+        item = assignment()
+
+        _ensure_assignment_shape(item)  # type: ignore[arg-type]
+        _ensure_assignment_source_rules(  # type: ignore[arg-type]
+            item,
+            function=function(),
+            base=None,
+            account_linked_to_identity=lambda _identity_id, _account_id: False,
+        )
+
+    def test_shape_rejects_backwards_validity_window(self) -> None:
+        now = datetime.now(timezone.utc)
+        item = assignment(valid_from=now, valid_until=now - timedelta(days=1))
+
+        self.assert_invalid("Valid until must be after valid from.", lambda: _ensure_assignment_shape(item))  # type: ignore[arg-type]
+
+    def test_shape_rejects_self_delegation(self) -> None:
+        item = assignment(id="assignment-1", delegated_from_assignment_id="assignment-1")
+
+        self.assert_invalid("A function assignment cannot delegate from itself.", lambda: _ensure_assignment_shape(item))  # type: ignore[arg-type]
+
+    def test_delegated_assignment_accepts_different_target(self) -> None:
+        base = assignment(id="source-1", identity_id="identity-1", account_id="account-1")
+        item = assignment(
+            id="assignment-2",
+            identity_id="identity-2",
+            account_id="account-2",
+            source="delegated",
+            delegated_from_assignment_id="source-1",
+        )
+
+        _ensure_assignment_source_rules(  # type: ignore[arg-type]
+            item,
+            function=function(delegable=True),
+            base=base,
+            account_linked_to_identity=lambda _identity_id, _account_id: False,
+        )
+
+    def test_delegated_assignment_rejects_same_target(self) -> None:
+        base = assignment(id="source-1", identity_id="identity-1", account_id="account-1")
+        item = assignment(source="delegated", delegated_from_assignment_id="source-1")
+
+        self.assert_invalid(
+            "A delegated assignment must target another identity or account.",
+            lambda: _ensure_assignment_source_rules(  # type: ignore[arg-type]
+                item,
+                function=function(delegable=True),
+                base=base,
+                account_linked_to_identity=lambda _identity_id, _account_id: False,
+            ),
+        )
+
+    def test_acting_for_assignment_accepts_identity_linked_account(self) -> None:
+        base = assignment(id="source-1", identity_id="identity-1", account_id=None)
+        item = assignment(
+            id="assignment-2",
+            source="acting_for",
+            delegated_from_assignment_id="source-1",
+            account_id="acting-account",
+            acting_for_account_id="represented-account",
+        )
+
+        _ensure_assignment_source_rules(  # type: ignore[arg-type]
+            item,
+            function=function(act_in_place_allowed=True),
+            base=base,
+            account_linked_to_identity=lambda identity_id, account_id: identity_id == "identity-1" and account_id == "represented-account",
+        )
+
+    def test_acting_for_assignment_rejects_unlinked_represented_account(self) -> None:
+        base = assignment(id="source-1", identity_id="identity-1", account_id=None)
+        item = assignment(
+            id="assignment-2",
+            source="acting_for",
+            delegated_from_assignment_id="source-1",
+            account_id="acting-account",
+            acting_for_account_id="represented-account",
+        )
+
+        self.assert_invalid(
+            "Acting-for account must belong to the source assignment identity.",
+            lambda: _ensure_assignment_source_rules(  # type: ignore[arg-type]
+                item,
+                function=function(act_in_place_allowed=True),
+                base=base,
+                account_linked_to_identity=lambda _identity_id, _account_id: False,
+            ),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

webui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/idm-webui",
-  "version": "0.1.7",
+  "version": "0.1.8",
   "private": true,
   "type": "module",
   "main": "src/index.ts",
@@ -14,7 +14,7 @@
     "./styles/idm.css": "./src/styles/idm.css"
   },
   "peerDependencies": {
-    "@govoplan/core-webui": "^0.1.7",
+    "@govoplan/core-webui": "^0.1.8",
     "@vitejs/plugin-react": "^4.3.4",
     "lucide-react": "^1.23.0",
     "react": "^19.0.0",