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