mirror of
https://github.com/suitenumerique/messages.git
synced 2026-08-17 21:25:41 +02:00
✨(api) add dynamic abilities field to MailDomain API
- Add get_abilities() method to MailDomain model for permission-based abilities - Update MailDomainAdminSerializer to inherit from AbilitiesModelSerializer - Add RetrieveModelMixin to MailDomainAdminViewSet for detail endpoints - Optimize queries with JOIN and annotation for regular users - Add prefetch_related for superusers to avoid N+1 queries - Add comprehensive tests for abilities functionality and query optimization - Add new test file for MailDomain model abilities The abilities field dynamically controls user permissions based on their role in the mail domain, with optimized database queries to maintain performance.
This commit is contained in:
@@ -500,8 +500,8 @@ class MailboxAccessWriteSerializer(serializers.ModelSerializer):
|
||||
return attrs
|
||||
|
||||
|
||||
class MailDomainAdminSerializer(serializers.ModelSerializer):
|
||||
"""Serialize MailDomain basic information for admin listing."""
|
||||
class MailDomainAdminSerializer(AbilitiesModelSerializer):
|
||||
"""Serialize mail domains for admin view."""
|
||||
|
||||
class Meta:
|
||||
model = models.MailDomain
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
"""Admin ViewSets for MailDomain and Mailbox management."""
|
||||
|
||||
from django.conf import settings
|
||||
from django.db.models import Q
|
||||
from django.db.models import F, Q
|
||||
from django.shortcuts import get_object_or_404
|
||||
from django.utils.translation import gettext_lazy as _ # For user-facing error messages
|
||||
|
||||
@@ -29,7 +29,9 @@ from core.api import serializers as core_serializers
|
||||
from core.identity.keycloak import reset_keycloak_user_password
|
||||
|
||||
|
||||
class AdminMailDomainViewSet(mixins.ListModelMixin, viewsets.GenericViewSet):
|
||||
class AdminMailDomainViewSet(
|
||||
mixins.ListModelMixin, viewsets.GenericViewSet, mixins.RetrieveModelMixin
|
||||
):
|
||||
"""
|
||||
ViewSet for listing MailDomains the user administers.
|
||||
Provides a top-level entry for mail domain administration.
|
||||
@@ -45,11 +47,20 @@ class AdminMailDomainViewSet(mixins.ListModelMixin, viewsets.GenericViewSet):
|
||||
return models.MailDomain.objects.none()
|
||||
|
||||
if user.is_superuser and user.is_staff:
|
||||
return models.MailDomain.objects.all().order_by("name")
|
||||
|
||||
return models.MailDomain.objects.filter(
|
||||
accesses__user=user, accesses__role=models.MailDomainAccessRoleChoices.ADMIN
|
||||
).order_by("name")
|
||||
# For superusers, preload accesses to avoid N+1 queries in get_abilities
|
||||
return models.MailDomain.objects.prefetch_related("accesses").order_by(
|
||||
"name"
|
||||
)
|
||||
# Optimization : one query with JOIN and annotation
|
||||
return (
|
||||
models.MailDomain.objects.filter(
|
||||
accesses__user=user,
|
||||
accesses__role=models.MailDomainAccessRoleChoices.ADMIN,
|
||||
)
|
||||
.annotate(user_role=F("accesses__role"))
|
||||
.distinct()
|
||||
.order_by("name")
|
||||
)
|
||||
|
||||
|
||||
class AdminMailDomainMailboxViewSet(
|
||||
|
||||
@@ -258,6 +258,44 @@ class MailDomain(BaseModel):
|
||||
]
|
||||
return records
|
||||
|
||||
def get_abilities(self, user):
|
||||
"""
|
||||
Compute and return abilities for a given user on the mail domain.
|
||||
"""
|
||||
role = None
|
||||
|
||||
if user.is_authenticated:
|
||||
try:
|
||||
role = self.user_role
|
||||
except AttributeError:
|
||||
# Use prefetched accesses if available to avoid additional queries
|
||||
if (
|
||||
hasattr(self, "_prefetched_objects_cache")
|
||||
and "accesses" in self._prefetched_objects_cache
|
||||
):
|
||||
# Find the user's access in the prefetched accesses
|
||||
for access in self.accesses.all():
|
||||
if access.user_id == user.id:
|
||||
role = access.role
|
||||
break
|
||||
else:
|
||||
try:
|
||||
role = self.accesses.filter(user=user).values("role")[0]["role"]
|
||||
except (MailDomainAccess.DoesNotExist, IndexError):
|
||||
role = None
|
||||
|
||||
is_admin = role == MailDomainAccessRoleChoices.ADMIN
|
||||
|
||||
return {
|
||||
"get": bool(role),
|
||||
"patch": is_admin,
|
||||
"put": is_admin,
|
||||
"post": is_admin,
|
||||
"delete": is_admin,
|
||||
"manage_accesses": is_admin,
|
||||
"manage_mailboxes": is_admin,
|
||||
}
|
||||
|
||||
|
||||
class Mailbox(BaseModel):
|
||||
"""Mailbox model to store mailbox information."""
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
"""Tests for the MailDomain Admin API endpoints."""
|
||||
# pylint: disable=unused-argument
|
||||
# pylint: disable=redefined-outer-name, unused-argument
|
||||
|
||||
from django.urls import reverse
|
||||
|
||||
@@ -122,10 +122,12 @@ class TestAdminMailDomainViewSet:
|
||||
mail_domain1,
|
||||
mail_domain2,
|
||||
unmanaged_domain,
|
||||
django_assert_num_queries,
|
||||
):
|
||||
"""Test that a domain admin can list domains they have admin access to."""
|
||||
api_client.force_authenticate(user=domain_admin_user)
|
||||
response = api_client.get(self.LIST_DOMAINS_URL)
|
||||
with django_assert_num_queries(2): # 1 for list + 1 for pagination
|
||||
response = api_client.get(self.LIST_DOMAINS_URL)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.data["count"] == 2
|
||||
@@ -226,3 +228,149 @@ class TestAdminMailDomainViewSet:
|
||||
assert str(mail_domain1.id) in domain_ids
|
||||
assert str(mail_domain2.id) not in domain_ids
|
||||
assert str(unmanaged_domain.id) not in domain_ids
|
||||
|
||||
def test_list_administered_maildomains_query_optimization(
|
||||
self,
|
||||
api_client,
|
||||
domain_admin_user,
|
||||
django_assert_num_queries,
|
||||
):
|
||||
"""Test that the query optimization works with multiple maildomains."""
|
||||
# Create several maildomains with access
|
||||
maildomains = []
|
||||
for i in range(5):
|
||||
maildomain = factories.MailDomainFactory(name=f"domain{i}.com")
|
||||
models.MailDomainAccess.objects.create(
|
||||
maildomain=maildomain,
|
||||
user=domain_admin_user,
|
||||
role=models.MailDomainAccessRoleChoices.ADMIN,
|
||||
)
|
||||
maildomains.append(maildomain)
|
||||
|
||||
# Create some maildomains without access
|
||||
for i in range(3):
|
||||
factories.MailDomainFactory(name=f"noaccess{i}.com")
|
||||
|
||||
api_client.force_authenticate(user=domain_admin_user)
|
||||
|
||||
with django_assert_num_queries(2): # 1 for list + 1 for pagination
|
||||
response = api_client.get(self.LIST_DOMAINS_URL)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.data["count"] == 5
|
||||
|
||||
# Verify that all maildomains with access are present
|
||||
domain_ids = [item["id"] for item in response.data["results"]]
|
||||
for maildomain in maildomains:
|
||||
assert str(maildomain.id) in domain_ids
|
||||
|
||||
def test_list_administered_maildomains_superuser_query_optimization(
|
||||
self,
|
||||
api_client,
|
||||
django_assert_num_queries,
|
||||
):
|
||||
"""Test that superuser query is also optimized."""
|
||||
# Create several maildomains
|
||||
maildomains = []
|
||||
for i in range(10):
|
||||
maildomain = factories.MailDomainFactory(name=f"domain{i}.com")
|
||||
maildomains.append(maildomain)
|
||||
|
||||
superuser = factories.UserFactory(is_superuser=True, is_staff=True)
|
||||
api_client.force_authenticate(user=superuser)
|
||||
with django_assert_num_queries(
|
||||
3
|
||||
): # 1 for list + 1 for pagination + 1 for abilities
|
||||
response = api_client.get(self.LIST_DOMAINS_URL)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.data["count"] == 10
|
||||
|
||||
def test_maildomain_retrieve_query_optimization(
|
||||
self,
|
||||
api_client,
|
||||
domain_admin_user,
|
||||
domain_admin_access1,
|
||||
mail_domain1,
|
||||
django_assert_num_queries,
|
||||
):
|
||||
"""Test that maildomain retrieve endpoint is optimized for queries."""
|
||||
api_client.force_authenticate(user=domain_admin_user)
|
||||
|
||||
with django_assert_num_queries(
|
||||
1
|
||||
): # 1 query to retrieve maildomain with annotation
|
||||
response = api_client.get(f"{self.LIST_DOMAINS_URL}{mail_domain1.id}/")
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.data["id"] == str(mail_domain1.id)
|
||||
|
||||
|
||||
class TestMailDomainAbilitiesAPI:
|
||||
"""Test the abilities field in MailDomain API responses."""
|
||||
|
||||
def test_maildomain_abilities_in_response(
|
||||
self, api_client, domain_admin_user, domain_admin_access1
|
||||
):
|
||||
"""Test that abilities are included in mail domain API response."""
|
||||
api_client.force_authenticate(user=domain_admin_user)
|
||||
url = reverse(
|
||||
"admin-maildomains-detail", args=[domain_admin_access1.maildomain.id]
|
||||
)
|
||||
response = api_client.get(url)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert "abilities" in response.data
|
||||
abilities = response.data["abilities"]
|
||||
assert abilities["get"] is True
|
||||
assert abilities["patch"] is True
|
||||
assert abilities["put"] is True
|
||||
assert abilities["post"] is True
|
||||
assert abilities["delete"] is True
|
||||
assert abilities["manage_accesses"] is True
|
||||
assert abilities["manage_mailboxes"] is True
|
||||
|
||||
def test_maildomain_list_with_abilities(
|
||||
self, api_client, domain_admin_user, domain_admin_access1
|
||||
):
|
||||
"""Test that mail domain list includes abilities for each domain."""
|
||||
api_client.force_authenticate(user=domain_admin_user)
|
||||
url = reverse("admin-maildomains-list")
|
||||
response = api_client.get(url)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert len(response.data["results"]) == 1
|
||||
|
||||
domain_data = response.data["results"][0]
|
||||
assert "abilities" in domain_data
|
||||
abilities = domain_data["abilities"]
|
||||
assert abilities["get"] is True
|
||||
assert abilities["patch"] is True
|
||||
assert abilities["put"] is True
|
||||
assert abilities["post"] is True
|
||||
assert abilities["delete"] is True
|
||||
assert abilities["manage_accesses"] is True
|
||||
assert abilities["manage_mailboxes"] is True
|
||||
|
||||
def test_maildomain_detail_no_access_abilities(
|
||||
self, api_client, other_user, mail_domain1
|
||||
):
|
||||
"""Test that abilities are correctly set when user has no access to detail."""
|
||||
api_client.force_authenticate(user=other_user)
|
||||
url = reverse("admin-maildomains-detail", args=[mail_domain1.id])
|
||||
response = api_client.get(url)
|
||||
|
||||
# Should return 404 since user has no access to this domain
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
def test_maildomain_list_no_access_abilities(
|
||||
self, api_client, other_user, mail_domain1, mail_domain2
|
||||
):
|
||||
"""Test that abilities are correctly set when user has no access."""
|
||||
api_client.force_authenticate(user=other_user)
|
||||
url = reverse("admin-maildomains-list")
|
||||
response = api_client.get(url)
|
||||
|
||||
# User has no access, so should get empty list
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert len(response.data["results"]) == 0
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
"""Tests for the MailDomain permissions system based on get_abilities."""
|
||||
# pylint: disable=redefined-outer-name,unused-argument
|
||||
|
||||
import pytest
|
||||
|
||||
from core import models
|
||||
from core.factories import MailDomainFactory, UserFactory
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def user():
|
||||
"""Create a test user."""
|
||||
return UserFactory()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def maildomain():
|
||||
"""Create a test mail domain."""
|
||||
return MailDomainFactory()
|
||||
|
||||
|
||||
class TestMailDomainModelAbilities:
|
||||
"""Test the get_abilities methods on MailDomain models."""
|
||||
|
||||
def test_maildomain_get_abilities_no_access(self, user, maildomain):
|
||||
"""Test MailDomain.get_abilities when user has no access."""
|
||||
abilities = maildomain.get_abilities(user)
|
||||
|
||||
assert abilities["get"] is False
|
||||
assert abilities["patch"] is False
|
||||
assert abilities["put"] is False
|
||||
assert abilities["post"] is False
|
||||
assert abilities["delete"] is False
|
||||
assert abilities["manage_accesses"] is False
|
||||
assert abilities["manage_mailboxes"] is False
|
||||
|
||||
def test_maildomain_get_abilities_admin(self, user, maildomain):
|
||||
"""Test MailDomain.get_abilities when user has admin access."""
|
||||
models.MailDomainAccess.objects.create(
|
||||
maildomain=maildomain,
|
||||
user=user,
|
||||
role=models.MailDomainAccessRoleChoices.ADMIN,
|
||||
)
|
||||
|
||||
abilities = maildomain.get_abilities(user)
|
||||
|
||||
assert abilities["get"] is True
|
||||
assert abilities["patch"] is True
|
||||
assert abilities["put"] is True
|
||||
assert abilities["post"] is True
|
||||
assert abilities["delete"] is True
|
||||
assert abilities["manage_accesses"] is True
|
||||
assert abilities["manage_mailboxes"] is True
|
||||
Reference in New Issue
Block a user