feat: add canonical identity administration

tests/test_admin_api.py

@@ -0,0 +1,156 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+import shutil
+import tempfile
+import unittest
+from unittest.mock import patch
+from pathlib import Path
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from govoplan_core.auth import ApiPrincipal, get_api_principal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.db.base import Base
+from govoplan_core.db.session import configure_database, reset_database
+from govoplan_identity.backend.api.v1.routes import router
+
+
+class IdentityAdminApiTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.root = Path(tempfile.mkdtemp(prefix="govoplan-identity-admin-api-"))
+        self.database = configure_database(f"sqlite:///{self.root / 'identity.db'}")
+        Base.metadata.create_all(self.database.engine)
+        app = FastAPI()
+        app.include_router(router, prefix="/api/v1")
+        app.dependency_overrides[get_api_principal] = self._principal
+        self.client = TestClient(app)
+        self.audit_patches = (
+            patch("govoplan_identity.backend.api.v1.routes.audit_from_principal"),
+            patch("govoplan_identity.backend.lifecycle.audit_event"),
+        )
+        self.route_audit = self.audit_patches[0].start()
+        self.lifecycle_audit = self.audit_patches[1].start()
+
+    def tearDown(self) -> None:
+        self.client.close()
+        for item in reversed(self.audit_patches):
+            item.stop()
+        reset_database(dispose=True)
+        shutil.rmtree(self.root, ignore_errors=True)
+
+    def test_primary_lifecycle_and_system_scope_are_enforced(self) -> None:
+        created = self.client.post(
+            "/api/v1/identity/identities",
+            json={
+                "display_name": "Ada Example",
+                "external_subject": "subject-ada",
+                "source": "local",
+            },
+        )
+        self.assertEqual(201, created.status_code, created.text)
+        identity = created.json()
+        self.assertEqual("system", identity["management_scope"])
+        identity_id = identity["id"]
+
+        first = self.client.post(
+            f"/api/v1/identity/identities/{identity_id}/account-links",
+            json={"account_id": "account-1", "source": "local"},
+        )
+        self.assertEqual(201, first.status_code, first.text)
+        self.assertEqual("account-1", first.json()["primary_account_id"])
+
+        second = self.client.post(
+            f"/api/v1/identity/identities/{identity_id}/account-links",
+            json={"account_id": "account-2", "source": "idm:accepted"},
+        )
+        self.assertEqual(201, second.status_code, second.text)
+        links = second.json()["account_links"]
+        first_link = next(item for item in links if item["account_id"] == "account-1")
+        second_link = next(item for item in links if item["account_id"] == "account-2")
+
+        blocked = self.client.delete(
+            f"/api/v1/identity/identities/{identity_id}/account-links/{first_link['id']}"
+        )
+        self.assertEqual(409, blocked.status_code, blocked.text)
+
+        promoted = self.client.patch(
+            f"/api/v1/identity/identities/{identity_id}/account-links/{second_link['id']}",
+            json={"is_primary": True, "reason": "Preferred institutional account"},
+        )
+        self.assertEqual(200, promoted.status_code, promoted.text)
+        self.assertEqual("account-2", promoted.json()["primary_account_id"])
+
+        removed = self.client.delete(
+            f"/api/v1/identity/identities/{identity_id}/account-links/{first_link['id']}"
+        )
+        self.assertEqual(204, removed.status_code, removed.text)
+
+        deactivated = self.client.post(
+            f"/api/v1/identity/identities/{identity_id}/deactivate",
+            json={"reason": "Duplicate subject under review"},
+        )
+        self.assertEqual(200, deactivated.status_code, deactivated.text)
+        self.assertEqual("inactive", deactivated.json()["status"])
+
+        default_list = self.client.get("/api/v1/identity/identities")
+        self.assertEqual([], default_list.json()["identities"])
+        inclusive_list = self.client.get(
+            "/api/v1/identity/identities",
+            params={"include_inactive": "true"},
+        )
+        self.assertEqual(identity_id, inclusive_list.json()["identities"][0]["id"])
+        self.assertEqual("system", inclusive_list.json()["management_scope"])
+        self.assertEqual("tenant-1", inclusive_list.json()["tenant_context_id"])
+
+        self.assertTrue(
+            all(call.kwargs["scope"] == "system" for call in self.route_audit.call_args_list)
+        )
+        self.assertTrue(
+            all(call.kwargs["scope"] == "system" for call in self.lifecycle_audit.call_args_list)
+        )
+
+    def test_account_can_only_be_linked_to_one_identity(self) -> None:
+        identity_ids = []
+        for name in ("Ada", "Grace"):
+            response = self.client.post(
+                "/api/v1/identity/identities",
+                json={"display_name": name},
+            )
+            self.assertEqual(201, response.status_code, response.text)
+            identity_ids.append(response.json()["id"])
+
+        first = self.client.post(
+            f"/api/v1/identity/identities/{identity_ids[0]}/account-links",
+            json={"account_id": "account-shared"},
+        )
+        self.assertEqual(201, first.status_code, first.text)
+        conflict = self.client.post(
+            f"/api/v1/identity/identities/{identity_ids[1]}/account-links",
+            json={"account_id": "account-shared"},
+        )
+        self.assertEqual(409, conflict.status_code, conflict.text)
+
+    @staticmethod
+    def _principal() -> ApiPrincipal:
+        return ApiPrincipal(
+            principal=PrincipalRef(
+                account_id="account-admin",
+                membership_id="user-admin",
+                tenant_id="tenant-1",
+                scopes=frozenset(
+                    {
+                        "identity:identity:read",
+                        "identity:identity:admin",
+                        "identity:account_link:admin",
+                    }
+                ),
+            ),
+            account=SimpleNamespace(id="account-admin"),
+            user=SimpleNamespace(id="user-admin"),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()