add-ideas/govoplan-idm
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

intermittent commit

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-14 13:22:11 +02:00
parent b47f71916d
commit 74652686ca
2 changed files with 218 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
src/govoplan_idm/backend/api/v1/routes.py
View File

@@ -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,16 +178,45 @@ 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:
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":
_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:
@@ -183,21 +225,30 @@ def _ensure_assignment_workflow(
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":
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:
if item.acting_for_account_id != base.account_id:
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.")
elif not _account_linked_to_identity(session, base.identity_id, item.acting_for_account_id):
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.")
else:
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:

132
tests/test_assignment_workflow.py Normal file
View File

@@ -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()