145 lines
5.9 KiB
Python
145 lines
5.9 KiB
Python
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,
|
|
TENANT_EVENT_ERASURE_COMPLETED,
|
|
TENANT_EVENT_RESUMED,
|
|
TENANT_EVENT_SUSPENDED,
|
|
TENANT_LIFECYCLE_EVENTS,
|
|
tenant_lifecycle_event,
|
|
tenant_lifecycle_event_type,
|
|
)
|
|
|
|
|
|
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"))
|
|
self.assertEqual("tenant.suspended", tenant_lifecycle_event_type("suspended"))
|
|
self.assertEqual("tenant.resumed", tenant_lifecycle_event_type("resumed"))
|
|
self.assertEqual("tenant.deletion_requested", tenant_lifecycle_event_type("deletion_requested"))
|
|
self.assertEqual("tenant.erasure_completed", tenant_lifecycle_event_type("erasure_completed"))
|
|
self.assertEqual(
|
|
(
|
|
TENANT_EVENT_CREATED,
|
|
TENANT_EVENT_SUSPENDED,
|
|
TENANT_EVENT_RESUMED,
|
|
TENANT_EVENT_DELETION_REQUESTED,
|
|
TENANT_EVENT_ERASURE_COMPLETED,
|
|
),
|
|
TENANT_LIFECYCLE_EVENTS,
|
|
)
|
|
|
|
def test_lifecycle_event_builds_audit_details_without_losing_extra_fields(self) -> None:
|
|
occurred_at = datetime(2026, 7, 11, 12, 30, tzinfo=UTC)
|
|
|
|
event = tenant_lifecycle_event(
|
|
"deletion_requested",
|
|
tenant_id="tenant-1",
|
|
tenant_slug="city",
|
|
tenant_name="City Office",
|
|
actor_account_id="account-1",
|
|
requested_by_account_id="account-2",
|
|
reason="end of contract",
|
|
counts={"files": 2},
|
|
occurred_at=occurred_at,
|
|
details={"mode": "retire"},
|
|
)
|
|
|
|
self.assertEqual(TENANT_EVENT_DELETION_REQUESTED, event.event_type)
|
|
self.assertEqual("tenant-1", event.tenant_id)
|
|
self.assertEqual(
|
|
{
|
|
"mode": "retire",
|
|
"slug": "city",
|
|
"name": "City Office",
|
|
"actor_account_id": "account-1",
|
|
"requested_by_account_id": "account-2",
|
|
"reason": "end of contract",
|
|
"counts": {"files": 2},
|
|
"occurred_at": "2026-07-11T12:30:00+00:00",
|
|
},
|
|
event.audit_details(),
|
|
)
|
|
|
|
|
|
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()
|