From 8dd5123aab14ba3038a5298fab44764bc5819d2a Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 10 Jul 2026 18:57:40 +0200 Subject: [PATCH] Cover identity and organization API cleanup --- .../core/configuration_safety.py | 14 ++ tests/test_identity_api.py | 101 +++++++++++ tests/test_organizations_api.py | 168 ++++++++++++++++++ 3 files changed, 283 insertions(+) create mode 100644 tests/test_identity_api.py create mode 100644 tests/test_organizations_api.py diff --git a/src/govoplan_core/core/configuration_safety.py b/src/govoplan_core/core/configuration_safety.py index ce03ddf..cbb14e1 100644 --- a/src/govoplan_core/core/configuration_safety.py +++ b/src/govoplan_core/core/configuration_safety.py @@ -193,6 +193,20 @@ _CONFIGURATION_FIELD_SAFETY: tuple[ConfigurationFieldSafety, ...] = ( rollback_history_required=True, notes="Central groups, roles, and assignments need previews before materialization.", ), + ConfigurationFieldSafety( + key="organizations.model", + label="Organization model", + owner_module="organizations", + scope="tenant", + storage="module_settings", + ui_managed=True, + risk="high", + dry_run_required=True, + policy_explanation_required=True, + audit_event="organizations.model.updated", + rollback_history_required=True, + notes="Organization meta-model, concrete units, relations, functions, and assignments can be governed per tenant.", + ), ConfigurationFieldSafety( key="mail_profiles.credentials", label="Mail profile credentials", diff --git a/tests/test_identity_api.py b/tests/test_identity_api.py new file mode 100644 index 0000000..ce18f33 --- /dev/null +++ b/tests/test_identity_api.py @@ -0,0 +1,101 @@ +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.db.base import Base +from govoplan_identity.backend.api.v1.routes import router as identity_router +from govoplan_identity.backend.db.models import Identity, IdentityAccountLink +from tests.db_isolation import temporary_database + + +class IdentityApiTests(unittest.TestCase): + def _principal(self, scopes: set[str]) -> ApiPrincipal: + account = Account( + id="account-identity-api", + email="identity-admin@example.test", + normalized_email="identity-admin@example.test", + display_name="Identity Admin", + ) + user = User( + id="user-identity-api", + tenant_id="tenant-identity-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(identity_router, prefix="/api/v1") + app.dependency_overrides[get_api_principal] = lambda: self._principal(scopes) + return app + + def test_identity_search_returns_account_links(self) -> None: + root = Path(tempfile.mkdtemp(prefix="govoplan-identity-api-")) + try: + with temporary_database(f"sqlite:///{root / 'identity.db'}") as database: + Base.metadata.create_all(bind=database.engine) + with database.session() as session: + identity = Identity( + id="identity-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-alice-primary", + is_primary=True, + source="local", + ) + ) + session.add( + IdentityAccountLink( + identity_id=identity.id, + account_id="account-alice-secondary", + is_primary=False, + source="local", + ) + ) + session.commit() + + app = self._app({"identity:identity:read"}) + with TestClient(app) as client: + response = client.get("/api/v1/identity/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-alice", item["id"]) + self.assertEqual("Alice Registry", item["display_name"]) + self.assertEqual("account-alice-primary", item["primary_account_id"]) + self.assertEqual(["account-alice-primary", "account-alice-secondary"], item["account_ids"]) + finally: + shutil.rmtree(root, ignore_errors=True) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_organizations_api.py b/tests/test_organizations_api.py new file mode 100644 index 0000000..5bdb650 --- /dev/null +++ b/tests/test_organizations_api.py @@ -0,0 +1,168 @@ +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_organizations.backend.api.v1.routes import router as organizations_router +from tests.db_isolation import temporary_database + + +class OrganizationsApiTests(unittest.TestCase): + def _principal(self, scopes: set[str]) -> ApiPrincipal: + account = Account( + id="account-organizations-api", + email="org-admin@example.test", + normalized_email="org-admin@example.test", + display_name="Organization Admin", + ) + user = User( + id="user-organizations-api", + tenant_id="tenant-organizations-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(organizations_router, prefix="/api/v1") + app.dependency_overrides[get_api_principal] = lambda: self._principal(scopes) + return app + + def test_organization_settings_defaults_and_updates_are_tenant_scoped(self) -> None: + root = Path(tempfile.mkdtemp(prefix="govoplan-organizations-api-")) + try: + with temporary_database(f"sqlite:///{root / 'organizations.db'}") as database: + Base.metadata.create_all(bind=database.engine) + app = self._app({"organizations:settings:read", "organizations:settings:write"}) + + with TestClient(app) as client: + default_response = client.get("/api/v1/organizations/settings") + self.assertEqual(200, default_response.status_code, default_response.text) + self.assertIsNone(default_response.json()["id"]) + self.assertTrue(default_response.json()["allow_tenant_model_customization"]) + self.assertEqual("standard", default_response.json()["audit_detail_level"]) + + update_response = client.patch( + "/api/v1/organizations/settings", + json={ + "allow_tenant_model_customization": False, + "require_model_change_requests": True, + "audit_detail_level": "full", + "change_retention_days": 365, + }, + ) + self.assertEqual(200, update_response.status_code, update_response.text) + payload = update_response.json() + self.assertFalse(payload["allow_tenant_model_customization"]) + self.assertTrue(payload["require_model_change_requests"]) + self.assertEqual("full", payload["audit_detail_level"]) + self.assertEqual(365, payload["change_retention_days"]) + + reload_response = client.get("/api/v1/organizations/settings") + self.assertEqual(200, reload_response.status_code, reload_response.text) + self.assertEqual(payload["id"], reload_response.json()["id"]) + self.assertFalse(reload_response.json()["allow_tenant_model_customization"]) + finally: + shutil.rmtree(root, ignore_errors=True) + + def test_unit_parent_update_rejects_cycles(self) -> None: + root = Path(tempfile.mkdtemp(prefix="govoplan-organizations-cycle-")) + try: + with temporary_database(f"sqlite:///{root / 'organizations.db'}") as database: + Base.metadata.create_all(bind=database.engine) + app = self._app({"organizations:unit:write"}) + + with TestClient(app) as client: + root_response = client.post("/api/v1/organizations/units", json={"name": "Registry"}) + self.assertEqual(201, root_response.status_code, root_response.text) + root_id = root_response.json()["id"] + + child_response = client.post( + "/api/v1/organizations/units", + json={"name": "Registry Frontdesk", "parent_id": root_id}, + ) + self.assertEqual(201, child_response.status_code, child_response.text) + child_id = child_response.json()["id"] + + cycle_response = client.patch( + f"/api/v1/organizations/units/{root_id}", + json={"parent_id": child_id}, + ) + self.assertEqual(422, cycle_response.status_code, cycle_response.text) + self.assertIn("cycle", cycle_response.text) + finally: + shutil.rmtree(root, ignore_errors=True) + + def test_required_change_request_policy_blocks_and_allows_unit_writes(self) -> None: + root = Path(tempfile.mkdtemp(prefix="govoplan-organizations-change-request-")) + try: + with temporary_database(f"sqlite:///{root / 'organizations.db'}") as database: + Base.metadata.create_all(bind=database.engine) + app = self._app({"organizations:settings:read", "organizations:settings:write", "organizations:unit:write"}) + + with TestClient(app) as client: + settings_response = client.patch( + "/api/v1/organizations/settings", + json={"require_model_change_requests": True}, + ) + self.assertEqual(200, settings_response.status_code, settings_response.text) + + blocked_response = client.post("/api/v1/organizations/units", json={"name": "Registry"}) + self.assertEqual(409, blocked_response.status_code, blocked_response.text) + self.assertEqual("configuration_request_required", blocked_response.json()["detail"]["code"]) + + value = {"resource_type": "unit", "operation": "created", "payload": {"name": "Registry"}} + with database.session() as session: + request = create_configuration_change_request( + session, + key="organizations.model", + value=value, + actor_user_id="user-organizations-api", + actor_scopes=("organizations:unit:write",), + dry_run=True, + target={"tenant_id": "tenant-organizations-api", "resource_type": "unit", "operation": "created"}, + reason="Test organization unit creation", + ) + session.commit() + + allowed_response = client.post( + "/api/v1/organizations/units", + json={"name": "Registry", "change_request_id": request["id"]}, + ) + self.assertEqual(201, allowed_response.status_code, allowed_response.text) + self.assertEqual("Registry", allowed_response.json()["name"]) + + with database.session() as session: + snapshot = configuration_control_snapshot(session) + stored_request = next(item for item in snapshot["requests"] if item["id"] == request["id"]) + self.assertEqual("applied", stored_request["status"]) + self.assertTrue(any(item["approval_request_id"] == request["id"] for item in snapshot["history"])) + finally: + shutil.rmtree(root, ignore_errors=True) + + +if __name__ == "__main__": + unittest.main()