126 lines
5.8 KiB
Python
126 lines
5.8 KiB
Python
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()
|