jcv-dev
a2930426b4
- infrasynth.configs: typed multi-tenant config store (registry, service, secrets, cache) + public config_changed/config_reset signals and API - emit the declared-but-dead signals (features flags/overrides, scheduler task_completed/task_failed, tenancy tenant_updated, audit model_changed) and per-model audit field exclusions - security: permission catalog (security_permission), Django-style model-derived AutoPermission, PermissionRegistry, RoleAssignment, global-or-tenant Grant/Revoke, catalog API - consolidate the permission surface: PermissionRegistry only (drop the settings dict), IsAuthenticatedAndPermitted aliases HybridPermission, require_permission replaced by required_permissions + require_all - packaging: add [build-system]; add Forgejo publish workflow (.forgejo)
61 lines
2.2 KiB
Python
61 lines
2.2 KiB
Python
"""``tenant_updated`` is emitted from the tenant edit path (Part B3)."""
|
|
|
|
import pytest
|
|
|
|
from infrasynth.tenancy.services import TenantService
|
|
from infrasynth.tenancy.signals import tenant_updated
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
TENANTS_URL = "/api/v1/tenancy/tenants/"
|
|
|
|
|
|
def _capture(signal):
|
|
received = []
|
|
receiver = lambda **kwargs: received.append(kwargs) # noqa: E731
|
|
signal.connect(receiver, weak=False)
|
|
return received, receiver
|
|
|
|
|
|
class TestTenantUpdatedSignal:
|
|
def test_patch_emits_with_changed_fields(self, authenticated_client, tenant):
|
|
received, receiver = _capture(tenant_updated)
|
|
try:
|
|
response = authenticated_client.patch(
|
|
f"{TENANTS_URL}{tenant.pk}/", {"name": "Renamed", "locale": "en"}, format="json"
|
|
)
|
|
finally:
|
|
tenant_updated.disconnect(receiver)
|
|
assert response.status_code == 200
|
|
assert len(received) == 1
|
|
assert received[0]["tenant_id"] == str(tenant.pk)
|
|
assert set(received[0]["changes"]) == {"name", "locale"}
|
|
tenant.refresh_from_db()
|
|
assert tenant.name == "Renamed"
|
|
|
|
def test_no_signal_when_nothing_changes(self, authenticated_client, tenant):
|
|
received, receiver = _capture(tenant_updated)
|
|
try:
|
|
response = authenticated_client.patch(f"{TENANTS_URL}{tenant.pk}/", {}, format="json")
|
|
finally:
|
|
tenant_updated.disconnect(receiver)
|
|
assert response.status_code == 200
|
|
assert received == []
|
|
|
|
def test_service_update_emits(self, tenant):
|
|
received, receiver = _capture(tenant_updated)
|
|
try:
|
|
TenantService().update_tenant(tenant, timezone="Europe/Madrid")
|
|
finally:
|
|
tenant_updated.disconnect(receiver)
|
|
assert received[0]["changes"] == {"timezone": "Europe/Madrid"}
|
|
|
|
def test_service_update_ignores_disallowed_fields(self, tenant):
|
|
received, receiver = _capture(tenant_updated)
|
|
try:
|
|
TenantService().update_tenant(tenant, status="archived")
|
|
finally:
|
|
tenant_updated.disconnect(receiver)
|
|
assert received == []
|
|
tenant.refresh_from_db()
|
|
assert tenant.status == "active"
|