diff --git a/src/govoplan_idm/backend/api/v1/routes.py b/src/govoplan_idm/backend/api/v1/routes.py index 73dac9f..4b28e3a 100644 --- a/src/govoplan_idm/backend/api/v1/routes.py +++ b/src/govoplan_idm/backend/api/v1/routes.py @@ -1,6 +1,5 @@ from __future__ import annotations -from collections.abc import Callable from typing import Any, TypeVar from fastapi import APIRouter, Depends, HTTPException, Query, status @@ -39,6 +38,14 @@ from govoplan_core.core.organizations import ( from govoplan_core.core.runtime import get_registry from govoplan_core.db.session import get_session from govoplan_core.security.time import utc_now +from govoplan_idm.backend.assignment_transitions import ( + AssignmentMutationPlan, + AssignmentTransitionError, + lifecycle_event_types, + plan_assignment_update, + validate_assignment_shape, + validate_assignment_source_rules, +) from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment, IdmTenantSettings from .schemas import ( @@ -72,8 +79,6 @@ IDM_SETTINGS_READ_SCOPES = ( IDM_SETTINGS_WRITE_SCOPES = ("idm:settings:write",) IDM_ASSIGNMENT_CHANGE_CONTROL_KEY = "idm.organization_assignments" IDM_ASSIGNMENT_AUDIT_EVENT = "idm.organization_assignment.updated" -ASSIGNMENT_SOURCES = {"direct", "delegated", "acting_for", "directory", "governance", "system"} - ModelT = TypeVar("ModelT") @@ -114,6 +119,33 @@ def _commit(session: Session, item: ModelT) -> ModelT: return item +def _flush_assignment( + session: Session, + item: IdmOrganizationFunctionAssignment, +) -> None: + try: + session.flush() + except IntegrityError as exc: + session.rollback() + raise _conflict( + "The IDM assignment conflicts with an existing assignment." + ) from exc + + +def _commit_assignment_transaction( + session: Session, + item: IdmOrganizationFunctionAssignment, +) -> None: + try: + session.commit() + except IntegrityError as exc: + session.rollback() + raise _conflict( + "The IDM assignment conflicts with an existing assignment." + ) from exc + session.refresh(item) + + def _row_fields(item: object) -> dict[str, Any]: keys = [column.name for column in item.__table__.columns] # type: ignore[attr-defined] return {key: getattr(item, key) for key in keys} @@ -221,24 +253,46 @@ def _ensure_assignment_workflow( *, tenant_id: str, ) -> None: - _ensure_assignment_shape(item) - function = _organization_function(item.function_id, tenant_id) - base = _assignment_source_assignment(session, item, tenant_id=tenant_id) - _ensure_assignment_source_rules( - item, - function=function, - base=base, - account_linked_to_identity=_account_linked_to_identity, - ) + try: + validate_assignment_shape(item) + function = _organization_function(item.function_id, tenant_id) + base = _assignment_source_assignment(session, item, tenant_id=tenant_id) + validate_assignment_source_rules( + item, + function=function, + base=base, + account_linked_to_identity=_account_linked_to_identity, + ) + except AssignmentTransitionError as exc: + raise _invalid(str(exc)) from exc -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: - raise _invalid("Valid until must be after valid from.") - 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.") +def _plan_assignment_update( + session: Session, + item: IdmOrganizationFunctionAssignment, + payload: OrganizationFunctionAssignmentUpdateRequest, + *, + tenant_id: str, +) -> AssignmentMutationPlan: + values = payload.model_dump(exclude_unset=True, exclude={"change_request_id"}) + organization_unit_id: str | None = None + if "function_id" in values and values["function_id"] is not None: + function = _organization_function(str(values["function_id"]), tenant_id) + organization_unit_id = function.organization_unit_id + try: + plan = plan_assignment_update( + item, + values, + organization_unit_id=organization_unit_id, + ) + except AssignmentTransitionError as exc: + raise _invalid(str(exc)) from exc + + if plan.after.identity_id != plan.before.identity_id: + _ensure_identity(plan.after.identity_id) + _ensure_account_link(plan.after.identity_id, plan.after.account_id) + _ensure_assignment_workflow(session, plan.after, tenant_id=tenant_id) # type: ignore[arg-type] + return plan def _assignment_source_assignment( @@ -257,66 +311,6 @@ def _assignment_source_assignment( return base -def _ensure_assignment_source_rules( - item: IdmOrganizationFunctionAssignment, - *, - function: OrganizationFunctionRef, - 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: OrganizationFunctionRef, - 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: OrganizationFunctionRef, - 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: item = session.query(IdmTenantSettings).filter(IdmTenantSettings.tenant_id == tenant_id).one_or_none() return bool(item and item.require_assignment_change_requests) @@ -390,7 +384,6 @@ def _record_assignment_change_applied( target=target, audit_event=IDM_ASSIGNMENT_AUDIT_EVENT, ) - session.commit() def _audit_capability_available() -> bool: @@ -417,7 +410,7 @@ def _record_assignment_audit( object_type=resource_type, object_id=resource_id, details={"before": before, "after": after}, - commit=True, + commit=False, ) @@ -509,7 +502,8 @@ def update_idm_settings( if payload.settings is None: raise _invalid("Settings cannot be empty.") item.settings = payload.settings - result = _settings_item(_commit(session, item)) + session.flush() + result = _settings_item(item) _record_assignment_audit( session, principal, @@ -519,6 +513,8 @@ def update_idm_settings( before=before, after=result.model_dump(mode="json"), ) + session.commit() + session.refresh(item) return result @@ -584,16 +580,16 @@ def create_organization_function_assignment( _get_tenant_row(session, IdmOrganizationFunctionAssignment, item.delegated_from_assignment_id, tenant_id, "Delegated function assignment") _ensure_assignment_workflow(session, item, tenant_id=tenant_id) session.add(item) - saved = _commit(session, item) - result = _assignment_item(saved) + _flush_assignment(session, item) + result = _assignment_item(item) after = result.model_dump(mode="json") - _record_assignment_change_applied(session, principal, approval=approval, target={**target, "resource_id": saved.id}, before=None, after=after) + _record_assignment_change_applied(session, principal, approval=approval, target={**target, "resource_id": item.id}, before=None, after=after) _record_assignment_audit( session, principal, action="idm.organization_assignment.created", resource_type="idm_organization_function_assignment", - resource_id=saved.id, + resource_id=item.id, before=None, after=after, ) @@ -603,8 +599,8 @@ def create_organization_function_assignment( result, event_type="idm.function_assignment.created.v1", ) - session.commit() - return result + _commit_assignment_transaction(session, item) + return _assignment_item(item) @router.patch("/organization-function-assignments/{assignment_id}", response_model=OrganizationFunctionAssignmentItem) @@ -618,44 +614,15 @@ def update_organization_function_assignment( item = _get_tenant_row(session, IdmOrganizationFunctionAssignment, assignment_id, tenant_id, "Organization function assignment") before = _jsonable(_row_fields(item)) approval, target, _value = _ensure_assignment_change_allowed(session, principal, tenant_id=tenant_id, operation="updated", payload=payload, resource_id=assignment_id) - if "function_id" in payload.model_fields_set: - if payload.function_id is None: - raise _invalid("Function is required.") - function = _organization_function(payload.function_id, tenant_id) - item.function_id = function.id - item.organization_unit_id = function.organization_unit_id - identity_id = payload.identity_id if "identity_id" in payload.model_fields_set else item.identity_id - account_id = payload.account_id if "account_id" in payload.model_fields_set else item.account_id - if "identity_id" in payload.model_fields_set: - if payload.identity_id is None: - raise _invalid("Identity is required.") - _ensure_identity(payload.identity_id) - item.identity_id = payload.identity_id - _ensure_account_link(identity_id, account_id) - if "source" in payload.model_fields_set and payload.source is None: - raise _invalid("Assignment source is required.") - if "applies_to_subunits" in payload.model_fields_set and payload.applies_to_subunits is None: - raise _invalid("Subunit applicability cannot be empty.") - if "is_active" in payload.model_fields_set and payload.is_active is None: - raise _invalid("Active state cannot be empty.") - if "settings" in payload.model_fields_set and payload.settings is None: - raise _invalid("Settings cannot be empty.") - for field in ( - "account_id", - "applies_to_subunits", - "source", - "delegated_from_assignment_id", - "acting_for_account_id", - "valid_from", - "valid_until", - "is_active", - "settings", - ): - if field in payload.model_fields_set: - setattr(item, field, getattr(payload, field)) - _ensure_assignment_workflow(session, item, tenant_id=tenant_id) - saved = _commit(session, item) - result = _assignment_item(saved) + plan = _plan_assignment_update( + session, + item, + payload, + tenant_id=tenant_id, + ) + plan.apply(item) + _flush_assignment(session, item) + result = _assignment_item(item) after = result.model_dump(mode="json") _record_assignment_change_applied(session, principal, approval=approval, target=target, before=before, after=after) _record_assignment_audit( @@ -663,41 +630,23 @@ def update_organization_function_assignment( principal, action="idm.organization_assignment.updated", resource_type="idm_organization_function_assignment", - resource_id=saved.id, + resource_id=item.id, before=before, after=after, ) - _publish_assignment_event( - session, - principal, - result, - event_type="idm.function_assignment.changed.v1", - ) - before_values = before if isinstance(before, dict) else {} - if before_values.get("is_active") is True and result.is_active is False: - _publish_assignment_event( - session, - principal, - result, - event_type="idm.function_assignment.revoked.v1", - ) - previous_valid_until = before_values.get("valid_until") - if ( - result.valid_until is not None - and result.valid_until <= utc_now() - and ( - previous_valid_until is None - or str(previous_valid_until) != result.valid_until.isoformat() - ) + for event_type in lifecycle_event_types( + plan.before, + plan.after, + now=utc_now(), ): _publish_assignment_event( session, principal, result, - event_type="idm.function_assignment.expired.v1", + event_type=event_type, ) - session.commit() - return result + _commit_assignment_transaction(session, item) + return _assignment_item(item) @router.get("/organization-identities", response_model=OrganizationIdentityCandidateList) diff --git a/src/govoplan_idm/backend/assignment_transitions.py b/src/govoplan_idm/backend/assignment_transitions.py new file mode 100644 index 0000000..301be09 --- /dev/null +++ b/src/govoplan_idm/backend/assignment_transitions.py @@ -0,0 +1,295 @@ +from __future__ import annotations + +from collections.abc import Callable, Mapping +from dataclasses import dataclass, replace +from datetime import datetime, timezone +from typing import Any, Protocol + +from govoplan_core.core.organizations import OrganizationFunctionRef + + +ASSIGNMENT_SOURCES = { + "direct", + "delegated", + "acting_for", + "directory", + "governance", + "system", +} +MUTABLE_ASSIGNMENT_FIELDS = ( + "identity_id", + "account_id", + "function_id", + "applies_to_subunits", + "source", + "delegated_from_assignment_id", + "acting_for_account_id", + "valid_from", + "valid_until", + "is_active", + "settings", +) +REQUIRED_UPDATE_FIELDS = { + "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.", +} + + +class AssignmentTransitionError(ValueError): + pass + + +class AssignmentLike(Protocol): + id: str + identity_id: str + account_id: str | None + function_id: str + organization_unit_id: str + applies_to_subunits: bool + source: str + delegated_from_assignment_id: str | None + acting_for_account_id: str | None + valid_from: datetime | None + valid_until: datetime | None + is_active: bool + settings: dict[str, Any] + + +@dataclass(frozen=True, slots=True) +class AssignmentSnapshot: + id: str + identity_id: str + account_id: str | None + function_id: str + organization_unit_id: str + applies_to_subunits: bool + source: str + delegated_from_assignment_id: str | None + acting_for_account_id: str | None + valid_from: datetime | None + valid_until: datetime | None + is_active: bool + settings: dict[str, Any] + + @classmethod + def from_assignment(cls, item: AssignmentLike) -> AssignmentSnapshot: + return cls( + id=item.id, + identity_id=item.identity_id, + account_id=item.account_id, + function_id=item.function_id, + organization_unit_id=item.organization_unit_id, + applies_to_subunits=item.applies_to_subunits, + source=item.source, + delegated_from_assignment_id=item.delegated_from_assignment_id, + acting_for_account_id=item.acting_for_account_id, + valid_from=item.valid_from, + valid_until=item.valid_until, + is_active=item.is_active, + settings=dict(item.settings or {}), + ) + + +@dataclass(frozen=True, slots=True) +class AssignmentMutationPlan: + before: AssignmentSnapshot + after: AssignmentSnapshot + changed_fields: tuple[str, ...] + + def apply(self, item: AssignmentLike) -> None: + for field in self.changed_fields: + setattr(item, field, getattr(self.after, field)) + + +def plan_assignment_update( + item: AssignmentLike, + values: Mapping[str, object], + *, + organization_unit_id: str | None = None, +) -> AssignmentMutationPlan: + before = AssignmentSnapshot.from_assignment(item) + updates: dict[str, object] = {} + for field in MUTABLE_ASSIGNMENT_FIELDS: + if field not in values: + continue + value = values[field] + message = REQUIRED_UPDATE_FIELDS.get(field) + if value is None and message is not None: + raise AssignmentTransitionError(message) + updates[field] = value + if "function_id" in updates: + if organization_unit_id is None: + raise AssignmentTransitionError( + "The organization unit for the selected function is required." + ) + updates["organization_unit_id"] = organization_unit_id + if "settings" in updates: + updates["settings"] = dict(updates["settings"] or {}) # type: ignore[arg-type] + + after = replace(before, **updates) + validate_assignment_shape(after) + changed_fields = tuple( + field + for field in (*MUTABLE_ASSIGNMENT_FIELDS, "organization_unit_id") + if getattr(before, field) != getattr(after, field) + ) + return AssignmentMutationPlan( + before=before, + after=after, + changed_fields=changed_fields, + ) + + +def validate_assignment_shape(item: AssignmentLike) -> None: + if item.source not in ASSIGNMENT_SOURCES: + raise AssignmentTransitionError("Assignment source is not supported.") + if ( + item.valid_from is not None + and item.valid_until is not None + and _comparable_datetime(item.valid_until) + <= _comparable_datetime(item.valid_from) + ): + raise AssignmentTransitionError("Valid until must be after valid from.") + if ( + item.delegated_from_assignment_id is not None + and item.delegated_from_assignment_id == item.id + ): + raise AssignmentTransitionError( + "A function assignment cannot delegate from itself." + ) + + +def validate_assignment_source_rules( + item: AssignmentLike, + *, + function: OrganizationFunctionRef, + base: AssignmentLike | None, + account_linked_to_identity: Callable[[str, str], bool], +) -> None: + if item.source == "delegated": + _validate_delegated_assignment(item, function=function, base=base) + return + if item.source == "acting_for": + _validate_acting_for_assignment( + item, + function=function, + base=base, + account_linked_to_identity=account_linked_to_identity, + ) + return + if item.delegated_from_assignment_id is not None: + raise AssignmentTransitionError( + "Only delegated or acting-for assignments can reference a source assignment." + ) + if item.acting_for_account_id is not None: + raise AssignmentTransitionError( + "Only acting-for assignments can set an acting-for account." + ) + + +def lifecycle_event_types( + before: AssignmentSnapshot, + after: AssignmentSnapshot, + *, + now: datetime, +) -> tuple[str, ...]: + events = ["idm.function_assignment.changed.v1"] + if before.is_active and not after.is_active: + events.append("idm.function_assignment.revoked.v1") + if ( + after.valid_until is not None + and _comparable_datetime(after.valid_until) <= _comparable_datetime(now) + and before.valid_until != after.valid_until + ): + events.append("idm.function_assignment.expired.v1") + return tuple(events) + + +def _validate_delegated_assignment( + item: AssignmentLike, + *, + function: OrganizationFunctionRef, + base: AssignmentLike | None, +) -> None: + if base is None: + raise AssignmentTransitionError( + "Delegated assignments require a source assignment." + ) + if not function.delegable: + raise AssignmentTransitionError( + "This organization function does not allow delegation." + ) + if item.acting_for_account_id is not None: + raise AssignmentTransitionError( + "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 AssignmentTransitionError( + "A delegated assignment must target another identity or account." + ) + + +def _validate_acting_for_assignment( + item: AssignmentLike, + *, + function: OrganizationFunctionRef, + base: AssignmentLike | None, + account_linked_to_identity: Callable[[str, str], bool], +) -> None: + if base is None: + raise AssignmentTransitionError( + "Acting-for assignments require a source assignment." + ) + if not function.act_in_place_allowed: + raise AssignmentTransitionError( + "This organization function does not allow acting in place." + ) + if item.acting_for_account_id is None: + raise AssignmentTransitionError( + "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 AssignmentTransitionError( + "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 AssignmentTransitionError( + "Acting-for account must belong to the source assignment identity." + ) + if item.account_id == item.acting_for_account_id: + raise AssignmentTransitionError( + "The acting account and acting-for account must be different." + ) + + +def _comparable_datetime(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +__all__ = [ + "ASSIGNMENT_SOURCES", + "AssignmentMutationPlan", + "AssignmentSnapshot", + "AssignmentTransitionError", + "lifecycle_event_types", + "plan_assignment_update", + "validate_assignment_shape", + "validate_assignment_source_rules", +] diff --git a/tests/test_assignment_workflow.py b/tests/test_assignment_workflow.py index 1b6d1ce..c8f4e38 100644 --- a/tests/test_assignment_workflow.py +++ b/tests/test_assignment_workflow.py @@ -4,9 +4,14 @@ 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 +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: @@ -15,11 +20,15 @@ def assignment(**overrides: object) -> SimpleNamespace: "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) @@ -33,15 +42,15 @@ def function(**overrides: object) -> SimpleNamespace: class AssignmentWorkflowTests(unittest.TestCase): def assert_invalid(self, expected_detail: str, callback: object) -> None: - with self.assertRaises(HTTPException) as captured: + with self.assertRaises(AssignmentTransitionError) as captured: callback() - self.assertEqual(captured.exception.detail, expected_detail) + self.assertEqual(str(captured.exception), 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] + validate_assignment_shape(item) # type: ignore[arg-type] + validate_assignment_source_rules( # type: ignore[arg-type] item, function=function(), base=None, @@ -52,12 +61,12 @@ class AssignmentWorkflowTests(unittest.TestCase): 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] + 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: _ensure_assignment_shape(item)) # type: ignore[arg-type] + 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") @@ -69,7 +78,7 @@ class AssignmentWorkflowTests(unittest.TestCase): delegated_from_assignment_id="source-1", ) - _ensure_assignment_source_rules( # type: ignore[arg-type] + validate_assignment_source_rules( # type: ignore[arg-type] item, function=function(delegable=True), base=base, @@ -82,7 +91,7 @@ class AssignmentWorkflowTests(unittest.TestCase): self.assert_invalid( "A delegated assignment must target another identity or account.", - lambda: _ensure_assignment_source_rules( # type: ignore[arg-type] + lambda: validate_assignment_source_rules( # type: ignore[arg-type] item, function=function(delegable=True), base=base, @@ -100,7 +109,7 @@ class AssignmentWorkflowTests(unittest.TestCase): acting_for_account_id="represented-account", ) - _ensure_assignment_source_rules( # type: ignore[arg-type] + validate_assignment_source_rules( # type: ignore[arg-type] item, function=function(act_in_place_allowed=True), base=base, @@ -119,7 +128,7 @@ class AssignmentWorkflowTests(unittest.TestCase): self.assert_invalid( "Acting-for account must belong to the source assignment identity.", - lambda: _ensure_assignment_source_rules( # type: ignore[arg-type] + lambda: validate_assignment_source_rules( # type: ignore[arg-type] item, function=function(act_in_place_allowed=True), base=base, @@ -127,6 +136,103 @@ class AssignmentWorkflowTests(unittest.TestCase): ), ) + 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()