intermittent commit

2026-07-14 13:22:12 +02:00
2 changed files with 147 additions and 35 deletions
--- a/src/govoplan_tenancy/backend/api/v1/routes.py
+++ b/src/govoplan_tenancy/backend/api/v1/routes.py
@@ -70,6 +70,16 @@ TENANT_SETTINGS_RESOURCE = "tenant_settings_section"
 ADMIN_MODULE_ID = "admin"
 ADMIN_SYSTEM_SETTINGS_COLLECTION = "admin.system_settings"
 TENANT_SETTINGS_SECTIONS = ("identity", "locale", "languages", "settings")
+TENANT_NON_STATUS_UPDATE_FIELDS = {
+    "name",
+    "description",
+    "default_locale",
+    "settings",
+    "allow_custom_groups",
+    "allow_custom_roles",
+    "allow_api_keys",
+}
+TENANT_GOVERNANCE_OVERRIDE_FIELDS = ("allow_custom_groups", "allow_custom_roles", "allow_api_keys")


 def _tenant_access_provisioner() -> TenantAccessProvisioner:
@@ -97,6 +107,20 @@ def _require_permission(principal: ApiPrincipal, scope: str) -> None:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {scope}")


+def _require_tenant_update_permissions(principal: ApiPrincipal, payload: TenantUpdateRequest) -> None:
+    if payload.model_fields_set.intersection(TENANT_NON_STATUS_UPDATE_FIELDS):
+        _require_permission(principal, "system:tenants:update")
+    if payload.is_active is not None:
+        _require_permission(principal, "system:tenants:suspend")
+
+
+def _tenant_or_404(session: Session, tenant_id: str) -> Tenant:
+    tenant = session.get(Tenant, tenant_id)
+    if tenant is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tenant not found")
+    return tenant
+
+
 def _tenant_item(session: Session, tenant: Tenant) -> TenantAdminItem:
    governance = effective_tenant_governance(session, tenant)
    return TenantAdminItem(
@@ -492,6 +516,50 @@ def create_tenant(
    return _tenant_item(session, tenant)


+def _apply_tenant_content_updates(tenant: Tenant, payload: TenantUpdateRequest) -> None:
+    if payload.name is not None:
+        tenant.name = payload.name.strip()
+    if "description" in payload.model_fields_set:
+        tenant.description = _normalized_optional_text(payload.description)
+    if payload.default_locale is not None:
+        tenant.default_locale = _normalized_tenant_locale(payload.default_locale)
+    if payload.settings is not None:
+        tenant.settings = payload.settings
+
+
+def _normalized_optional_text(value: str | None) -> str | None:
+    if value is None:
+        return None
+    clean = value.strip()
+    return clean or None
+
+
+def _normalized_tenant_locale(value: str) -> str:
+    return value.strip() or "en"
+
+
+def _apply_tenant_governance_updates(session: Session, tenant: Tenant, payload: TenantUpdateRequest) -> None:
+    try:
+        for field in TENANT_GOVERNANCE_OVERRIDE_FIELDS:
+            if field in payload.model_fields_set:
+                value = getattr(payload, field)
+                assert_tenant_governance_override_allowed(session, field=field, value=value)
+                setattr(tenant, field, value)
+    except AdminValidationError as exc:
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
+
+
+def _apply_tenant_status_update(tenant: Tenant, payload: TenantUpdateRequest, principal: ApiPrincipal) -> None:
+    if payload.is_active is None:
+        return
+    if not payload.is_active and tenant.id == principal.tenant_id:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Switch to another tenant before suspending the active tenant.",
+        )
+    tenant.is_active = payload.is_active
+
+
@router.patch("/tenants/{tenant_id}", response_model=TenantAdminItem)
 def update_tenant(
    tenant_id: str,
@@ -499,43 +567,13 @@ def update_tenant(
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(get_api_principal),
 ):
-    non_status_fields = {"name", "description", "default_locale", "settings", "allow_custom_groups", "allow_custom_roles", "allow_api_keys"}
-    if payload.model_fields_set.intersection(non_status_fields):
-        _require_permission(principal, "system:tenants:update")
-    if payload.is_active is not None:
-        _require_permission(principal, "system:tenants:suspend")
-    tenant = session.get(Tenant, tenant_id)
-    if tenant is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tenant not found")
+    _require_tenant_update_permissions(principal, payload)
+    tenant = _tenant_or_404(session, tenant_id)
    was_active = tenant.is_active
    before_sections = _tenant_settings_sections(_tenant_settings_item(session, tenant))
-    if payload.name is not None:
-        tenant.name = payload.name.strip()
-    if "description" in payload.model_fields_set:
-        tenant.description = payload.description.strip() if payload.description and payload.description.strip() else None
-    if payload.default_locale is not None:
-        tenant.default_locale = payload.default_locale.strip() or "en"
-    if payload.settings is not None:
-        tenant.settings = payload.settings
-    try:
-        if "allow_custom_groups" in payload.model_fields_set:
-            assert_tenant_governance_override_allowed(session, field="allow_custom_groups", value=payload.allow_custom_groups)
-            tenant.allow_custom_groups = payload.allow_custom_groups
-        if "allow_custom_roles" in payload.model_fields_set:
-            assert_tenant_governance_override_allowed(session, field="allow_custom_roles", value=payload.allow_custom_roles)
-            tenant.allow_custom_roles = payload.allow_custom_roles
-        if "allow_api_keys" in payload.model_fields_set:
-            assert_tenant_governance_override_allowed(session, field="allow_api_keys", value=payload.allow_api_keys)
-            tenant.allow_api_keys = payload.allow_api_keys
-    except AdminValidationError as exc:
-        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
-    if payload.is_active is not None:
-        if not payload.is_active and tenant.id == principal.tenant_id:
-            raise HTTPException(
-                status_code=status.HTTP_409_CONFLICT,
-                detail="Switch to another tenant before suspending the active tenant.",
-            )
-        tenant.is_active = payload.is_active
+    _apply_tenant_content_updates(tenant, payload)
+    _apply_tenant_governance_updates(session, tenant, payload)
+    _apply_tenant_status_update(tenant, payload, principal)
    session.add(tenant)
    audit_event(
        session,
--- a/tests/test_tenant_lifecycle.py
+++ b/tests/test_tenant_lifecycle.py
@@ -2,7 +2,16 @@ from __future__ import annotations

 import unittest
 from datetime import UTC, datetime
+from types import SimpleNamespace

+from fastapi import HTTPException
+
+from govoplan_tenancy.backend.api.v1.routes import (
+    _apply_tenant_content_updates,
+    _apply_tenant_status_update,
+    _require_tenant_update_permissions,
+)
+from govoplan_tenancy.backend.api.v1.schemas import TenantUpdateRequest
 from govoplan_tenancy.backend.lifecycle import (
    TENANT_EVENT_CREATED,
    TENANT_EVENT_DELETION_REQUESTED,
@@ -15,6 +24,15 @@ from govoplan_tenancy.backend.lifecycle import (
 )


+class FakePrincipal:
+    def __init__(self, scopes: set[str], *, tenant_id: str = "tenant-1") -> None:
+        self.scopes = frozenset(scopes)
+        self.tenant_id = tenant_id
+
+    def has(self, required_scope: str) -> bool:
+        return required_scope in self.scopes
+
+
 class TenantLifecycleContractTests(unittest.TestCase):
    def test_lifecycle_event_names_are_stable(self) -> None:
        self.assertEqual("tenant.created", tenant_lifecycle_event_type("created"))
@@ -66,5 +84,61 @@ class TenantLifecycleContractTests(unittest.TestCase):
        )


+class TenantUpdateHelperTests(unittest.TestCase):
+    def test_tenant_update_permissions_separate_content_and_status_changes(self) -> None:
+        _require_tenant_update_permissions(
+            FakePrincipal({"system:tenants:suspend"}),  # type: ignore[arg-type]
+            TenantUpdateRequest(is_active=False),
+        )
+        _require_tenant_update_permissions(
+            FakePrincipal({"system:tenants:update"}),  # type: ignore[arg-type]
+            TenantUpdateRequest(name="Updated"),
+        )
+
+        with self.assertRaises(HTTPException) as missing_update:
+            _require_tenant_update_permissions(FakePrincipal(set()), TenantUpdateRequest(name="Updated"))  # type: ignore[arg-type]
+        self.assertEqual(403, missing_update.exception.status_code)
+        self.assertEqual("Missing scope: system:tenants:update", missing_update.exception.detail)
+
+        with self.assertRaises(HTTPException) as missing_suspend:
+            _require_tenant_update_permissions(FakePrincipal(set()), TenantUpdateRequest(is_active=False))  # type: ignore[arg-type]
+        self.assertEqual(403, missing_suspend.exception.status_code)
+        self.assertEqual("Missing scope: system:tenants:suspend", missing_suspend.exception.detail)
+
+    def test_tenant_content_updates_normalize_blank_fields_and_defaults(self) -> None:
+        tenant = SimpleNamespace(name="Old", description="Old description", default_locale="de", settings={})
+        payload = TenantUpdateRequest(
+            name="  New tenant  ",
+            description="   ",
+            default_locale="   ",
+            settings={"theme": "contrast"},
+        )
+
+        _apply_tenant_content_updates(tenant, payload)  # type: ignore[arg-type]
+
+        self.assertEqual("New tenant", tenant.name)
+        self.assertIsNone(tenant.description)
+        self.assertEqual("en", tenant.default_locale)
+        self.assertEqual({"theme": "contrast"}, tenant.settings)
+
+    def test_tenant_status_update_prevents_suspending_current_tenant(self) -> None:
+        tenant = SimpleNamespace(id="tenant-1", is_active=True)
+        principal = FakePrincipal({"system:tenants:suspend"}, tenant_id="tenant-1")
+
+        with self.assertRaises(HTTPException) as captured:
+            _apply_tenant_status_update(tenant, TenantUpdateRequest(is_active=False), principal)  # type: ignore[arg-type]
+
+        self.assertEqual(409, captured.exception.status_code)
+        self.assertEqual("Switch to another tenant before suspending the active tenant.", captured.exception.detail)
+
+    def test_tenant_status_update_allows_other_tenant_suspension(self) -> None:
+        tenant = SimpleNamespace(id="tenant-2", is_active=True)
+        principal = FakePrincipal({"system:tenants:suspend"}, tenant_id="tenant-1")
+
+        _apply_tenant_status_update(tenant, TenantUpdateRequest(is_active=False), principal)  # type: ignore[arg-type]
+
+        self.assertFalse(tenant.is_active)
+
+
 if __name__ == "__main__":
    unittest.main()