diff --git a/src/govoplan_idm/backend/api/v1/routes.py b/src/govoplan_idm/backend/api/v1/routes.py index f59a1b6..134b20e 100644 --- a/src/govoplan_idm/backend/api/v1/routes.py +++ b/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: diff --git a/tests/test_assignment_workflow.py b/tests/test_assignment_workflow.py new file mode 100644 index 0000000..1b6d1ce --- /dev/null +++ b/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()