From 1dde03854733c3228216daba7f007a6272c3099b Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 14 Jul 2026 13:22:12 +0200 Subject: [PATCH] intermittent commit --- src/govoplan_tenancy/backend/api/v1/routes.py | 108 ++++++++++++------ tests/test_tenant_lifecycle.py | 74 ++++++++++++ 2 files changed, 147 insertions(+), 35 deletions(-) diff --git a/src/govoplan_tenancy/backend/api/v1/routes.py b/src/govoplan_tenancy/backend/api/v1/routes.py index ca4bacb..6c7b5d3 100644 --- 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, diff --git a/tests/test_tenant_lifecycle.py b/tests/test_tenant_lifecycle.py index 6a80fc2..0dbba93 100644 --- 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()