(organization) add API endpoints

This provides a way to get information about
the organization and update their name for
administrators.
This commit is contained in:
Quentin BEY
2024-11-21 22:26:04 +01:00
committed by BEY Quentin
parent 6fe4818743
commit 5692c50f21
8 changed files with 284 additions and 2 deletions

View File

@@ -51,6 +51,24 @@ class DynamicFieldsModelSerializer(serializers.ModelSerializer):
self.fields.pop(field_name) self.fields.pop(field_name)
class OrganizationSerializer(serializers.ModelSerializer):
"""Serialize organizations."""
abilities = serializers.SerializerMethodField()
class Meta:
model = models.Organization
fields = ["id", "name", "registration_id_list", "domain_list", "abilities"]
read_only_fields = ["id", "registration_id_list", "domain_list"]
def get_abilities(self, organization) -> dict:
"""Return abilities of the logged-in user on the instance."""
request = self.context.get("request")
if request:
return organization.get_abilities(request.user)
return {}
class UserSerializer(DynamicFieldsModelSerializer): class UserSerializer(DynamicFieldsModelSerializer):
"""Serialize users.""" """Serialize users."""

View File

@@ -172,6 +172,45 @@ class ContactViewSet(
return super().perform_create(serializer) return super().perform_create(serializer)
class OrganizationViewSet(
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
viewsets.GenericViewSet,
):
"""
Organization ViewSet
GET /api/organizations/<organization_id>/
Return the organization details for the given id.
PUT /api/organizations/<organization_id>/
Update the organization details for the given id.
PATCH /api/organizations/<organization_id>/
Partially update the organization details for the given id.
"""
permission_classes = [permissions.AccessPermission]
queryset = models.Organization.objects.all()
serializer_class = serializers.OrganizationSerializer
throttle_classes = [BurstRateThrottle, SustainedRateThrottle]
def get_queryset(self):
"""Limit listed organizations to the one the user belongs to."""
return (
super()
.get_queryset()
.filter(pk=self.request.user.organization_id)
.annotate(
user_role=Subquery(
models.OrganizationAccess.objects.filter(
user=self.request.user, organization=OuterRef("pk")
).values("role")[:1]
)
)
)
class UserViewSet( class UserViewSet(
SerializerPerActionMixin, SerializerPerActionMixin,
mixins.UpdateModelMixin, mixins.UpdateModelMixin,

View File

@@ -138,6 +138,22 @@ class OrganizationFactory(factory.django.DjangoModelFactory):
) )
class OrganizationAccessFactory(factory.django.DjangoModelFactory):
"""Factory to create organization accesses for testing purposes."""
class Meta:
model = models.OrganizationAccess
user = factory.SubFactory(
"core.factories.UserFactory",
organization=factory.SelfAttribute("..organization"),
)
organization = factory.SubFactory(
"core.factories.OrganizationFactory", with_registration_id=True
)
role = factory.fuzzy.FuzzyChoice(models.OrganizationRoleChoices.values)
class UserFactory(factory.django.DjangoModelFactory): class UserFactory(factory.django.DjangoModelFactory):
"""A factory to create random users for testing purposes.""" """A factory to create random users for testing purposes."""

View File

@@ -40,7 +40,7 @@ with open(contact_schema_path, "r", encoding="utf-8") as contact_schema_file:
contact_schema = json.load(contact_schema_file) contact_schema = json.load(contact_schema_file)
class RoleChoices(models.TextChoices): class RoleChoices(models.TextChoices): # pylint: disable=too-many-ancestors
"""Defines the possible roles a user can have in a team.""" """Defines the possible roles a user can have in a team."""
MEMBER = "member", _("Member") MEMBER = "member", _("Member")
@@ -48,7 +48,7 @@ class RoleChoices(models.TextChoices):
OWNER = "owner", _("Owner") OWNER = "owner", _("Owner")
class OrganizationRoleChoices(models.TextChoices): class OrganizationRoleChoices(models.TextChoices): # pylint: disable=too-many-ancestors
""" """
Defines the possible roles a user can have in an organization. Defines the possible roles a user can have in an organization.
For now, we only have one role, but we might add more in the future. For now, we only have one role, but we might add more in the future.
@@ -355,6 +355,26 @@ class Organization(BaseModel):
"domain_list value must be unique across all instances." "domain_list value must be unique across all instances."
) )
def get_abilities(self, user):
"""
Compute and return abilities for a given user on the organization.
"""
try:
# Use the role from queryset annotation if available
is_admin = self.user_role == OrganizationRoleChoices.ADMIN
except AttributeError:
is_admin = self.organization_accesses.filter(
user=user,
role=OrganizationRoleChoices.ADMIN,
).exists()
return {
"get": user.organization_id == self.pk,
"patch": is_admin,
"put": is_admin,
"delete": False,
}
class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin): class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin):
"""User model to work with OIDC only authentication.""" """User model to work with OIDC only authentication."""

View File

@@ -0,0 +1 @@
"""Test organization API endpoints."""

View File

@@ -0,0 +1,91 @@
"""
Tests for Organizations API endpoint in People's core app: retrieve
"""
import pytest
from rest_framework import status
from core import factories
pytestmark = pytest.mark.django_db
def test_api_organizations_retrieve_anonymous(client):
"""Anonymous users should not be allowed to retrieve an organization."""
organization = factories.OrganizationFactory(with_registration_id=True)
response = client.get(f"/api/v1.0/organizations/{organization.pk}/")
assert response.status_code == status.HTTP_401_UNAUTHORIZED
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
def test_api_organizations_retrieve_authenticated_unrelated(client):
"""
Authenticated users should not be allowed to retrieve an organization to which they are
not related.
"""
user = factories.UserFactory()
organization = factories.OrganizationFactory(with_registration_id=True)
client.force_login(user)
response = client.get(
f"/api/v1.0/organizations/{organization.pk!s}/",
)
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json() == {"detail": "No Organization matches the given query."}
def test_api_organizations_retrieve_authenticated_belong_to_organization(client):
"""
Authenticated users should be allowed to retrieve an organization to which they
belong to.
"""
organization = factories.OrganizationFactory(
registration_id_list=["56618615316840", "31561861231231", "98781236231482"],
domain_list=["example.com", "example.org"],
)
user = factories.UserFactory(organization=organization)
client.force_login(user)
response = client.get(
f"/api/v1.0/organizations/{organization.pk!s}/",
)
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"id": str(organization.pk),
"name": organization.name,
"abilities": {"delete": False, "get": True, "patch": False, "put": False},
"domain_list": ["example.com", "example.org"],
"registration_id_list": ["56618615316840", "31561861231231", "98781236231482"],
}
def test_api_organizations_retrieve_authenticated_administrator(client):
"""
Authenticated users should be allowed to retrieve an organization
which they administrate.
"""
organization_access = (
factories.OrganizationAccessFactory()
) # only role is administrator for now
user = organization_access.user
organization = organization_access.organization
client.force_login(user)
response = client.get(
f"/api/v1.0/organizations/{organization.pk!s}/",
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["abilities"] == {
"delete": False,
"get": True,
"patch": True,
"put": True,
}

View File

@@ -0,0 +1,96 @@
"""
Tests for Organizations API endpoint in People's core app: update
"""
import pytest
from rest_framework import status
from core import factories
pytestmark = pytest.mark.django_db
def test_api_organizations_update_anonymous(client):
"""Anonymous users should not be allowed to update an organization."""
organization = factories.OrganizationFactory(with_registration_id=True)
response = client.patch(
f"/api/v1.0/organizations/{organization.pk}/",
{"name": "New Name"},
content_type="application/json",
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
def test_api_organizations_update_authenticated_unrelated(client):
"""
Authenticated users should not be allowed to update an organization to which they are
not related.
"""
user = factories.UserFactory()
organization = factories.OrganizationFactory(with_registration_id=True)
client.force_login(user)
response = client.patch(
f"/api/v1.0/organizations/{organization.pk}/",
{"name": "New Name"},
content_type="application/json",
)
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json() == {"detail": "No Organization matches the given query."}
def test_api_organizations_update_authenticated_belong_to_organization(client):
"""
Authenticated users should NOT be allowed to update an organization to which they
belong to.
"""
organization = factories.OrganizationFactory(with_registration_id=True)
user = factories.UserFactory(organization=organization)
client.force_login(user)
response = client.patch(
f"/api/v1.0/organizations/{organization.pk}/",
{"name": "New Name"},
content_type="application/json",
)
assert response.status_code == status.HTTP_403_FORBIDDEN
assert response.json() == {
"detail": "You do not have permission to perform this action."
}
def test_api_organizations_update_authenticated_administrator(client):
"""
Authenticated users should be allowed to update an organization
which they administrate.
"""
organization = factories.OrganizationFactory(
registration_id_list=["56618615316840", "31561861231231", "98781236231482"],
domain_list=["example.com", "example.org"],
)
organization_access = factories.OrganizationAccessFactory(organization=organization)
user = organization_access.user
client.force_login(user)
response = client.patch(
f"/api/v1.0/organizations/{organization.pk}/",
{"name": "New Name"},
content_type="application/json",
)
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"id": str(organization.pk),
"name": "New Name",
"abilities": {"delete": False, "get": True, "patch": True, "put": True},
"domain_list": ["example.com", "example.org"],
"registration_id_list": ["56618615316840", "31561861231231", "98781236231482"],
}

View File

@@ -12,6 +12,7 @@ from core.resource_server.urls import urlpatterns as resource_server_urls
# - Main endpoints # - Main endpoints
router = DefaultRouter() router = DefaultRouter()
router.register("contacts", viewsets.ContactViewSet, basename="contacts") router.register("contacts", viewsets.ContactViewSet, basename="contacts")
router.register("organizations", viewsets.OrganizationViewSet, basename="organizations")
router.register("teams", viewsets.TeamViewSet, basename="teams") router.register("teams", viewsets.TeamViewSet, basename="teams")
router.register("users", viewsets.UserViewSet, basename="users") router.register("users", viewsets.UserViewSet, basename="users")
router.register( router.register(