add-ideas/govoplan-tenancy
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

intermittent commit

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-14 13:22:12 +02:00
parent efbec82761
commit 1dde038547
2 changed files with 147 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
src/govoplan_tenancy/backend/api/v1/routes.py
View File

@@ -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,

74
tests/test_tenant_lifecycle.py
View File

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