from __future__ import annotations import shutil import tempfile import unittest from pathlib import Path from fastapi import FastAPI from fastapi.testclient import TestClient from govoplan_access.backend.db.models import Account, User from govoplan_core.auth import ApiPrincipal, get_api_principal from govoplan_core.core.access import PrincipalRef from govoplan_core.core.configuration_control import configuration_control_snapshot, create_configuration_change_request from govoplan_core.db.base import Base from govoplan_identity.backend.db.models import Identity, IdentityAccountLink from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment, IdmTenantSettings from govoplan_idm.backend.api.v1.routes import router as idm_router from govoplan_idm.backend.manifest import get_manifest from govoplan_organizations.backend.db.models import OrganizationFunction, OrganizationUnit from tests.db_isolation import temporary_database class IdmApiTests(unittest.TestCase): def _principal(self, scopes: set[str]) -> ApiPrincipal: account = Account( id="account-idm-api", email="idm-admin@example.test", normalized_email="idm-admin@example.test", display_name="IDM Admin", ) user = User( id="user-idm-api", tenant_id="tenant-idm-api", account_id=account.id, email=account.email, display_name=account.display_name, settings={}, mail_profile_policy={}, ) return ApiPrincipal( principal=PrincipalRef( account_id=account.id, membership_id=user.id, tenant_id=user.tenant_id, scopes=frozenset(scopes), ), account=account, user=user, ) def _app(self, scopes: set[str]) -> FastAPI: app = FastAPI() app.include_router(idm_router, prefix="/api/v1") app.dependency_overrides[get_api_principal] = lambda: self._principal(scopes) return app def test_idm_hard_depends_on_identity_and_organizations(self) -> None: manifest = get_manifest() self.assertEqual(("identity", "organizations"), manifest.dependencies) def test_organization_identity_candidates_are_idm_bridge(self) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-idm-api-")) try: with temporary_database(f"sqlite:///{root / 'idm.db'}") as database: Base.metadata.create_all(bind=database.engine) with database.session() as session: identity = Identity( id="identity-idm-alice", display_name="Alice Registry", external_subject="alice@example.test", source="local", ) session.add(identity) session.add( IdentityAccountLink( identity_id=identity.id, account_id="account-idm-alice-primary", is_primary=True, source="local", ) ) session.commit() app = self._app({"organizations:function:assign"}) with TestClient(app) as client: response = client.get("/api/v1/idm/organization-identities", params={"query": "registry"}) self.assertEqual(200, response.status_code, response.text) payload = response.json() self.assertEqual(1, len(payload["identities"])) item = payload["identities"][0] self.assertEqual("identity-idm-alice", item["id"]) self.assertEqual("Alice Registry", item["display_name"]) self.assertEqual("account-idm-alice-primary", item["primary_account_id"]) finally: shutil.rmtree(root, ignore_errors=True) def test_organization_function_assignments_are_owned_by_idm(self) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-idm-assignments-api-")) try: with temporary_database(f"sqlite:///{root / 'idm-assignments.db'}") as database: Base.metadata.create_all(bind=database.engine) with database.session() as session: identity = Identity( id="identity-idm-bob", display_name="Bob Registry", external_subject="bob@example.test", source="local", ) unit = OrganizationUnit( id="unit-idm-registry", tenant_id="tenant-idm-api", slug="registry", name="Registry", settings={}, ) function = OrganizationFunction( id="function-idm-clerk", tenant_id="tenant-idm-api", organization_unit_id=unit.id, slug="registry-clerk", name="Registry Clerk", settings={}, ) session.add_all([ identity, IdentityAccountLink( identity_id=identity.id, account_id="account-idm-bob-primary", is_primary=True, source="local", ), unit, function, ]) session.commit() app = self._app({"idm:organization_assignment:read", "idm:organization_assignment:write"}) with TestClient(app) as client: create_response = client.post( "/api/v1/idm/organization-function-assignments", json={ "identity_id": "identity-idm-bob", "account_id": "account-idm-bob-primary", "function_id": "function-idm-clerk", "applies_to_subunits": True, }, ) self.assertEqual(201, create_response.status_code, create_response.text) created = create_response.json() self.assertEqual("identity-idm-bob", created["identity_id"]) self.assertEqual("function-idm-clerk", created["function_id"]) self.assertEqual("unit-idm-registry", created["organization_unit_id"]) self.assertTrue(created["applies_to_subunits"]) list_response = client.get("/api/v1/idm/organization-function-assignments") self.assertEqual(200, list_response.status_code, list_response.text) self.assertEqual([created["id"]], [item["id"] for item in list_response.json()["assignments"]]) patch_response = client.patch( f"/api/v1/idm/organization-function-assignments/{created['id']}", json={"is_active": False, "applies_to_subunits": False}, ) self.assertEqual(200, patch_response.status_code, patch_response.text) patched = patch_response.json() self.assertFalse(patched["is_active"]) self.assertFalse(patched["applies_to_subunits"]) with database.session() as session: self.assertEqual(1, session.query(IdmOrganizationFunctionAssignment).count()) finally: shutil.rmtree(root, ignore_errors=True) def test_legacy_organization_assign_scope_remains_assignment_compatibility(self) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-idm-legacy-scope-")) try: with temporary_database(f"sqlite:///{root / 'idm-legacy.db'}") as database: Base.metadata.create_all(bind=database.engine) with database.session() as session: session.add(Identity(id="identity-idm-legacy", display_name="Legacy Person", source="local")) session.add(IdentityAccountLink(identity_id="identity-idm-legacy", account_id="account-idm-legacy", is_primary=True, source="local")) session.add(OrganizationUnit(id="unit-idm-legacy", tenant_id="tenant-idm-api", slug="legacy", name="Legacy", settings={})) session.add( OrganizationFunction( id="function-idm-legacy", tenant_id="tenant-idm-api", organization_unit_id="unit-idm-legacy", slug="legacy-clerk", name="Legacy Clerk", settings={}, ) ) session.commit() app = self._app({"organizations:function:assign"}) with TestClient(app) as client: create_response = client.post( "/api/v1/idm/organization-function-assignments", json={ "identity_id": "identity-idm-legacy", "account_id": "account-idm-legacy", "function_id": "function-idm-legacy", }, ) self.assertEqual(201, create_response.status_code, create_response.text) list_response = client.get("/api/v1/idm/organization-function-assignments") self.assertEqual(200, list_response.status_code, list_response.text) self.assertEqual([create_response.json()["id"]], [item["id"] for item in list_response.json()["assignments"]]) finally: shutil.rmtree(root, ignore_errors=True) def test_idm_assignment_change_requests_can_be_required_per_tenant(self) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-idm-change-control-")) try: with temporary_database(f"sqlite:///{root / 'idm-change-control.db'}") as database: Base.metadata.create_all(bind=database.engine) with database.session() as session: session.add(Identity(id="identity-idm-change", display_name="Change Person", source="local")) session.add(IdentityAccountLink(identity_id="identity-idm-change", account_id="account-idm-change", is_primary=True, source="local")) session.add(OrganizationUnit(id="unit-idm-change", tenant_id="tenant-idm-api", slug="change", name="Change", settings={})) session.add( OrganizationFunction( id="function-idm-change", tenant_id="tenant-idm-api", organization_unit_id="unit-idm-change", slug="change-clerk", name="Change Clerk", settings={}, ) ) session.add( IdmTenantSettings( tenant_id="tenant-idm-api", require_assignment_change_requests=True, audit_detail_level="standard", settings={}, ) ) session.commit() app = self._app({"idm:organization_assignment:write", "idm:organization_assignment:read"}) payload = { "identity_id": "identity-idm-change", "account_id": "account-idm-change", "function_id": "function-idm-change", "applies_to_subunits": True, } with TestClient(app) as client: blocked = client.post("/api/v1/idm/organization-function-assignments", json=payload) self.assertEqual(409, blocked.status_code, blocked.text) self.assertEqual("configuration_request_required", blocked.json()["detail"]["code"]) with database.session() as session: request = create_configuration_change_request( session, key="idm.organization_assignments", value={ "resource_type": "organization_function_assignment", "operation": "created", "payload": payload, }, actor_user_id="user-idm-api", actor_scopes=("idm:organization_assignment:write",), dry_run=True, target={ "tenant_id": "tenant-idm-api", "resource_type": "organization_function_assignment", "operation": "created", }, ) session.commit() with TestClient(app) as client: applied = client.post( "/api/v1/idm/organization-function-assignments", json={**payload, "change_request_id": request["id"]}, ) self.assertEqual(201, applied.status_code, applied.text) self.assertEqual("identity-idm-change", applied.json()["identity_id"]) with database.session() as session: history = configuration_control_snapshot(session)["history"] self.assertEqual("idm.organization_assignments", history[0]["key"]) self.assertEqual("idm.organization_assignment.updated", history[0]["audit_event"]) finally: shutil.rmtree(root, ignore_errors=True) def test_idm_settings_can_be_read_and_updated(self) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-idm-settings-")) try: with temporary_database(f"sqlite:///{root / 'idm-settings.db'}") as database: Base.metadata.create_all(bind=database.engine) app = self._app({"idm:settings:read", "idm:settings:write"}) with TestClient(app) as client: initial = client.get("/api/v1/idm/settings") self.assertEqual(200, initial.status_code, initial.text) self.assertFalse(initial.json()["require_assignment_change_requests"]) updated = client.patch( "/api/v1/idm/settings", json={ "require_assignment_change_requests": True, "audit_detail_level": "full", "change_retention_days": 90, }, ) self.assertEqual(200, updated.status_code, updated.text) payload = updated.json() self.assertTrue(payload["require_assignment_change_requests"]) self.assertEqual("full", payload["audit_detail_level"]) self.assertEqual(90, payload["change_retention_days"]) finally: shutil.rmtree(root, ignore_errors=True) def test_delegated_assignments_require_delegable_function_and_source_assignment(self) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-idm-delegation-rules-")) try: with temporary_database(f"sqlite:///{root / 'idm-delegation.db'}") as database: Base.metadata.create_all(bind=database.engine) with database.session() as session: session.add_all([ Identity(id="identity-idm-source", display_name="Source Person", source="local"), IdentityAccountLink(identity_id="identity-idm-source", account_id="account-idm-source", is_primary=True, source="local"), Identity(id="identity-idm-delegate", display_name="Delegate Person", source="local"), IdentityAccountLink(identity_id="identity-idm-delegate", account_id="account-idm-delegate", is_primary=True, source="local"), OrganizationUnit(id="unit-idm-delegation", tenant_id="tenant-idm-api", slug="delegation", name="Delegation", settings={}), OrganizationFunction( id="function-idm-not-delegable", tenant_id="tenant-idm-api", organization_unit_id="unit-idm-delegation", slug="not-delegable", name="Not Delegable", delegable=False, settings={}, ), OrganizationFunction( id="function-idm-delegable", tenant_id="tenant-idm-api", organization_unit_id="unit-idm-delegation", slug="delegable", name="Delegable", delegable=True, settings={}, ), ]) session.add( IdmOrganizationFunctionAssignment( id="assignment-idm-source", tenant_id="tenant-idm-api", identity_id="identity-idm-source", account_id="account-idm-source", function_id="function-idm-delegable", organization_unit_id="unit-idm-delegation", applies_to_subunits=False, source="direct", settings={}, ) ) session.commit() app = self._app({"idm:organization_assignment:read", "idm:organization_assignment:write"}) with TestClient(app) as client: missing_source = client.post( "/api/v1/idm/organization-function-assignments", json={ "identity_id": "identity-idm-delegate", "account_id": "account-idm-delegate", "function_id": "function-idm-delegable", "source": "delegated", }, ) self.assertEqual(422, missing_source.status_code, missing_source.text) not_delegable = client.post( "/api/v1/idm/organization-function-assignments", json={ "identity_id": "identity-idm-delegate", "account_id": "account-idm-delegate", "function_id": "function-idm-not-delegable", "source": "delegated", "delegated_from_assignment_id": "assignment-idm-source", }, ) self.assertEqual(422, not_delegable.status_code, not_delegable.text) delegated = client.post( "/api/v1/idm/organization-function-assignments", json={ "identity_id": "identity-idm-delegate", "account_id": "account-idm-delegate", "function_id": "function-idm-delegable", "source": "delegated", "delegated_from_assignment_id": "assignment-idm-source", }, ) self.assertEqual(201, delegated.status_code, delegated.text) self.assertEqual("delegated", delegated.json()["source"]) finally: shutil.rmtree(root, ignore_errors=True) if __name__ == "__main__": unittest.main()