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