239 lines
8.9 KiB
Python
239 lines
8.9 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
|
|
from govoplan_idm.backend.assignment_transitions import (
|
|
AssignmentSnapshot,
|
|
AssignmentTransitionError,
|
|
lifecycle_event_types,
|
|
plan_assignment_update,
|
|
validate_assignment_shape,
|
|
validate_assignment_source_rules,
|
|
)
|
|
|
|
|
|
def assignment(**overrides: object) -> SimpleNamespace:
|
|
values = {
|
|
"id": "assignment-1",
|
|
"identity_id": "identity-1",
|
|
"account_id": "account-1",
|
|
"function_id": "function-1",
|
|
"organization_unit_id": "unit-1",
|
|
"applies_to_subunits": False,
|
|
"source": "direct",
|
|
"delegated_from_assignment_id": None,
|
|
"acting_for_account_id": None,
|
|
"valid_from": None,
|
|
"valid_until": None,
|
|
"is_active": True,
|
|
"settings": {},
|
|
}
|
|
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(AssignmentTransitionError) as captured:
|
|
callback()
|
|
self.assertEqual(str(captured.exception), expected_detail)
|
|
|
|
def test_direct_assignment_accepts_plain_shape(self) -> None:
|
|
item = assignment()
|
|
|
|
validate_assignment_shape(item) # type: ignore[arg-type]
|
|
validate_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: validate_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: validate_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",
|
|
)
|
|
|
|
validate_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: validate_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",
|
|
)
|
|
|
|
validate_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: validate_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,
|
|
),
|
|
)
|
|
|
|
def test_update_plan_does_not_mutate_until_applied(self) -> None:
|
|
item = assignment()
|
|
|
|
plan = plan_assignment_update(
|
|
item, # type: ignore[arg-type]
|
|
{
|
|
"function_id": "function-2",
|
|
"account_id": "account-2",
|
|
"settings": {"source": "test"},
|
|
},
|
|
organization_unit_id="unit-2",
|
|
)
|
|
|
|
self.assertEqual("function-1", item.function_id)
|
|
self.assertEqual("function-2", plan.after.function_id)
|
|
self.assertEqual("unit-2", plan.after.organization_unit_id)
|
|
self.assertEqual(
|
|
("account_id", "function_id", "settings", "organization_unit_id"),
|
|
plan.changed_fields,
|
|
)
|
|
plan.apply(item) # type: ignore[arg-type]
|
|
self.assertEqual("function-2", item.function_id)
|
|
self.assertEqual("unit-2", item.organization_unit_id)
|
|
|
|
def test_update_decision_table_rejects_required_nulls(self) -> None:
|
|
item = assignment()
|
|
cases = {
|
|
"identity_id": "Identity is required.",
|
|
"function_id": "Function is required.",
|
|
"applies_to_subunits": "Subunit applicability cannot be empty.",
|
|
"source": "Assignment source is required.",
|
|
"is_active": "Active state cannot be empty.",
|
|
"settings": "Settings cannot be empty.",
|
|
}
|
|
for field, message in cases.items():
|
|
with self.subTest(field=field):
|
|
self.assert_invalid(
|
|
message,
|
|
lambda field=field: plan_assignment_update( # type: ignore[misc]
|
|
item, # type: ignore[arg-type]
|
|
{field: None},
|
|
organization_unit_id=(
|
|
"unit-2" if field == "function_id" else None
|
|
),
|
|
),
|
|
)
|
|
|
|
def test_lifecycle_event_decision_table(self) -> None:
|
|
now = datetime.now(timezone.utc)
|
|
before = AssignmentSnapshot.from_assignment(assignment()) # type: ignore[arg-type]
|
|
cases = (
|
|
(
|
|
"ordinary update",
|
|
plan_assignment_update(
|
|
assignment(), # type: ignore[arg-type]
|
|
{"settings": {"changed": True}},
|
|
).after,
|
|
("idm.function_assignment.changed.v1",),
|
|
),
|
|
(
|
|
"revocation",
|
|
plan_assignment_update(
|
|
assignment(), # type: ignore[arg-type]
|
|
{"is_active": False},
|
|
).after,
|
|
(
|
|
"idm.function_assignment.changed.v1",
|
|
"idm.function_assignment.revoked.v1",
|
|
),
|
|
),
|
|
(
|
|
"new expiration",
|
|
plan_assignment_update(
|
|
assignment(), # type: ignore[arg-type]
|
|
{"valid_until": now - timedelta(minutes=1)},
|
|
).after,
|
|
(
|
|
"idm.function_assignment.changed.v1",
|
|
"idm.function_assignment.expired.v1",
|
|
),
|
|
),
|
|
(
|
|
"future validity",
|
|
plan_assignment_update(
|
|
assignment(), # type: ignore[arg-type]
|
|
{"valid_until": now + timedelta(minutes=1)},
|
|
).after,
|
|
("idm.function_assignment.changed.v1",),
|
|
),
|
|
)
|
|
for label, after, expected in cases:
|
|
with self.subTest(label=label):
|
|
self.assertEqual(
|
|
expected,
|
|
lifecycle_event_types(before, after, now=now),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|