feat(identity): enforce lifecycle audit semantics

2026-08-19 23:19:10 +02:00
parent f4938c9666
commit 43f0128e6a
4 changed files with 345 additions and 0 deletions
@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from govoplan_core.db.base import Base
+from govoplan_core.db.session import configure_database, reset_database
+from govoplan_identity.backend.db.models import Identity, IdentityAccountLink
+from govoplan_identity.backend.directory import SqlIdentityDirectory
+from govoplan_identity.backend.lifecycle import (
+    IdentityLifecycleError,
+    set_identity_active,
+    set_primary_account,
+)
+
+
+class IdentityLifecycleTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.database = configure_database("sqlite:///:memory:")
+        Base.metadata.create_all(
+            self.database.engine,
+            tables=[Identity.__table__, IdentityAccountLink.__table__],
+        )
+        with self.database.session() as session:
+            session.add(Identity(id="identity-1", display_name="Ada", source="local", is_active=True, settings={}))
+            session.add_all(
+                [
+                    IdentityAccountLink(id="link-1", identity_id="identity-1", account_id="account-1", is_primary=True, source="local"),
+                    IdentityAccountLink(id="link-2", identity_id="identity-1", account_id="account-2", is_primary=False, source="idm:accepted"),
+                ]
+            )
+            session.commit()
+
+    def tearDown(self) -> None:
+        reset_database(dispose=True)
+
+    @patch("govoplan_identity.backend.lifecycle.audit_event")
+    def test_primary_account_change_preserves_all_links_and_audits_source(self, audit) -> None:
+        with self.database.session() as session:
+            result = set_primary_account(
+                session,
+                identity_id="identity-1",
+                account_id="account-2",
+                actor_tenant_id="tenant-1",
+                actor_user_id="user-1",
+                reason="Preferred institutional account",
+            )
+            session.commit()
+
+        self.assertTrue(result.changed)
+        self.assertEqual("account-1", result.previous_primary_account_id)
+        self.assertEqual("account-2", result.primary_account_id)
+        with self.database.session() as session:
+            links = session.query(IdentityAccountLink).filter_by(identity_id="identity-1").order_by(IdentityAccountLink.account_id).all()
+            self.assertEqual([False, True], [item.is_primary for item in links])
+        resolved = SqlIdentityDirectory().get_identity("identity-1")
+        self.assertIsNotNone(resolved)
+        self.assertEqual(("account-1", "account-2"), tuple(sorted(resolved.account_ids)))
+        self.assertEqual("account-2", resolved.primary_account_id)
+        self.assertEqual("identity.primary_account_changed", audit.call_args.kwargs["action"])
+        self.assertEqual("idm:accepted", audit.call_args.kwargs["details"]["link_source"])
+
+    @patch("govoplan_identity.backend.lifecycle.audit_event")
+    def test_deactivation_is_reversible_and_preserves_account_links(self, audit) -> None:
+        with self.database.session() as session:
+            changed = set_identity_active(
+                session,
+                identity_id="identity-1",
+                active=False,
+                actor_tenant_id="tenant-1",
+                actor_user_id="user-1",
+            )
+            session.commit()
+        self.assertTrue(changed.changed)
+        inactive = SqlIdentityDirectory().get_identity("identity-1")
+        self.assertIsNotNone(inactive)
+        self.assertEqual("inactive", inactive.status)
+        self.assertEqual((), SqlIdentityDirectory().search_identities())
+        with self.database.session() as session:
+            self.assertEqual(2, session.query(IdentityAccountLink).filter_by(identity_id="identity-1").count())
+            set_identity_active(
+                session,
+                identity_id="identity-1",
+                active=True,
+                actor_tenant_id="tenant-1",
+                actor_user_id="user-1",
+            )
+            session.commit()
+        self.assertIsNotNone(SqlIdentityDirectory().get_identity("identity-1"))
+        self.assertEqual(["identity.deactivated", "identity.activated"], [call.kwargs["action"] for call in audit.call_args_list])
+
+    @patch("govoplan_identity.backend.lifecycle.audit_event")
+    def test_invalid_primary_change_does_not_mutate_or_audit(self, audit) -> None:
+        with self.database.session() as session:
+            with self.assertRaises(IdentityLifecycleError) as raised:
+                set_primary_account(
+                    session,
+                    identity_id="identity-1",
+                    account_id="not-linked",
+                    actor_tenant_id="tenant-1",
+                    actor_user_id="user-1",
+                )
+            self.assertEqual("account_not_linked", raised.exception.code)
+            links = session.query(IdentityAccountLink).filter_by(identity_id="identity-1").order_by(IdentityAccountLink.account_id).all()
+            self.assertEqual([True, False], [item.is_primary for item in links])
+        audit.assert_not_called()
+
+    @patch("govoplan_identity.backend.lifecycle.audit_event", side_effect=RuntimeError("audit unavailable"))
+    def test_primary_change_rolls_back_when_audit_cannot_be_recorded(self, _audit) -> None:
+        with self.database.session() as session:
+            with self.assertRaisesRegex(RuntimeError, "audit unavailable"):
+                set_primary_account(
+                    session,
+                    identity_id="identity-1",
+                    account_id="account-2",
+                    actor_tenant_id="tenant-1",
+                    actor_user_id="user-1",
+                )
+            session.expire_all()
+            links = session.query(IdentityAccountLink).filter_by(identity_id="identity-1").order_by(IdentityAccountLink.account_id).all()
+            self.assertEqual([True, False], [item.is_primary for item in links])
+
+
+if __name__ == "__main__":
+    unittest.main()