157 lines
6.2 KiB
Python
157 lines
6.2 KiB
Python
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()
|