Cover identity and organization API cleanup

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()