Close the gaps between the documented contract (API-STANDARD, TENANCY, ENTITLEMENTS) and the implementation, and remove committed build artifacts. Security: - verify + process inbound webhooks (HMAC/handler verify, size limit, timestamp tolerance, idempotency via InboundEvent.external_id) - real 2FA login flow (pre-auth challenge; tokens only after verify/recovery) - wire HybridPermission into security/audit views; add API-key rotate and users/<id>/permissions|roles endpoints - tenant-scoped throttling on by default; webhook replay protection - verify MercadoPago webhook signatures - login brute-force guard, configurable password policy, real ALTCHA PoW Correctness: - apply verified billing webhooks idempotently (subscription/entitlement/ invoice/PaymentTransaction); scheduled payment lifecycle jobs - capture audit update diffs automatically; add audit retention purge - working notification retries, per-channel rate limits, log retention - pluggable virus scanner, upload-size limit, pipeline toggle - feature rollout %/environment targeting; settings-driven registrations - workflow guards (instance cap, route depth, self-assignment, clone on re-entry) - wire every previously-dead INFRASYNTH_* setting; drop truly dead ones Delivery: - README + CHANGELOG; CI format check + coverage gate - keep test media out of the tree; untrack .coverage, __pycache__, egg-info, docs/ and invoice artifacts
93 lines
3.7 KiB
Python
93 lines
3.7 KiB
Python
"""TenantService behaviour (TENANCY.md §3, §6, §8)."""
|
|
|
|
import pytest
|
|
from django.contrib.auth import get_user_model
|
|
|
|
from infrasynth.shared.exceptions import EntitlementError, NotFoundError, ValidationAppError
|
|
from infrasynth.tenancy.models import Tenant, TenantMembership
|
|
from infrasynth.tenancy.services import TenantService
|
|
|
|
UserModel = get_user_model()
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
class TestCreateTenant:
|
|
def test_creates_owner_membership(self, user):
|
|
tenant = TenantService().create_tenant("Acme Ltd", user)
|
|
assert tenant.slug == "acme-ltd"
|
|
assert TenantMembership.objects.filter(tenant=tenant, user=user, is_owner=True, role="owner").exists()
|
|
|
|
def test_slug_is_deduplicated(self, user):
|
|
svc = TenantService()
|
|
a = svc.create_tenant("Acme", user)
|
|
b = svc.create_tenant("Acme", user)
|
|
assert a.slug != b.slug
|
|
|
|
def test_requires_authenticated_owner(self):
|
|
with pytest.raises(ValidationAppError):
|
|
TenantService().create_tenant("Acme", None)
|
|
|
|
|
|
class TestSelectTenant:
|
|
def test_selects_active_membership(self, user, tenant):
|
|
assert TenantService().select_tenant(user, tenant.pk) == tenant
|
|
|
|
def test_cross_tenant_raises_not_found(self, user):
|
|
other = Tenant.objects.create(slug="elsewhere", name="Elsewhere")
|
|
with pytest.raises(NotFoundError):
|
|
TenantService().select_tenant(user, other.pk)
|
|
|
|
def test_suspended_tenant_denied(self, user, tenant):
|
|
tenant.status = Tenant.Status.SUSPENDED
|
|
tenant.save(update_fields=["status"])
|
|
with pytest.raises(EntitlementError):
|
|
TenantService().select_tenant(user, tenant.pk)
|
|
|
|
|
|
class TestLifecycle:
|
|
def test_suspend_then_reinstate(self, tenant):
|
|
svc = TenantService()
|
|
svc.suspend(tenant, reason="nonpayment")
|
|
tenant.refresh_from_db()
|
|
assert tenant.status == Tenant.Status.SUSPENDED
|
|
assert tenant.suspended_at is not None
|
|
|
|
svc.reinstate(tenant)
|
|
tenant.refresh_from_db()
|
|
assert tenant.status == Tenant.Status.ACTIVE
|
|
assert tenant.suspended_at is None
|
|
|
|
def test_offboard_archives_and_revokes(self, user, tenant):
|
|
TenantService().offboard(tenant)
|
|
tenant.refresh_from_db()
|
|
assert tenant.status == Tenant.Status.ARCHIVED
|
|
assert not TenantMembership.objects.filter(tenant=tenant, is_active=True).exists()
|
|
|
|
|
|
class TestMembership:
|
|
def test_remove_member(self, tenant):
|
|
other = UserModel.objects.create_user(username="m", password="x")
|
|
TenantMembership.objects.create(tenant=tenant, user=other)
|
|
assert TenantService().remove_member(tenant, other) is True
|
|
assert not TenantMembership.objects.filter(tenant=tenant, user=other, is_active=True).exists()
|
|
|
|
def test_invite_and_accept(self, tenant, user):
|
|
invitation = TenantService().invite(tenant, "new@example.com", invited_by=user)
|
|
assert invitation.token
|
|
new_user = UserModel.objects.create_user(username="new", email="new@example.com", password="x")
|
|
membership = TenantService().accept_invitation(invitation.token, new_user)
|
|
assert membership.tenant_id == tenant.id
|
|
invitation.refresh_from_db()
|
|
assert invitation.accepted_at is not None
|
|
|
|
def test_expired_invitation_rejected(self, tenant, user):
|
|
from datetime import timedelta
|
|
|
|
from django.utils import timezone
|
|
|
|
invitation = TenantService().invite(tenant, "late@example.com")
|
|
invitation.expires_at = timezone.now() - timedelta(days=1)
|
|
invitation.save(update_fields=["expires_at"])
|
|
new_user = UserModel.objects.create_user(username="late", password="x")
|
|
with pytest.raises(NotFoundError):
|
|
TenantService().accept_invitation(invitation.token, new_user)
|