Added permissions endpoint

Author: epou

api/serializers.py

@@ -1,5 +1,6 @@
+from abc import abstractmethod
 from datetime import datetime
-from typing import Literal, Optional
+from typing import Literal, Optional, Union
 from uuid import UUID
 
 from django.contrib.auth import get_user_model
@@ -11,6 +12,7 @@
 from rest_framework import serializers
 
 from drf_extra_fields.geo_fields import PointField
+from rest_framework_dataclasses.serializers import DataclassSerializer
 from taggit.serializers import TaggitSerializer
 
 from tigacrafting.models import (
@@ -21,6 +23,7 @@
     PhotoPrediction,
     FavoritedReports
 )
+from tigacrafting.permissions import Permissions
 from tigaserver_app.models import (
     NotificationContent,
     Notification,
@@ -102,6 +105,82 @@ class Meta:
             "received_at": {"source": "server_upload_time"},
         }
 
+class PermissionsSerializer(DataclassSerializer):
+    class Meta:
+        dataclass = Permissions
+
+class BaseRolePermissionSerializer(serializers.Serializer):
+    role = serializers.SerializerMethodField()
+    permissions = serializers.SerializerMethodField()
+
+    @abstractmethod
+    def _get_role(self, obj: User):
+        raise NotImplementedError()
+
+    def get_role(self, obj: Union[User, TigaUser]) -> Literal['base', 'annotator', 'supervisor', 'reviewer', 'admin']:
+        if isinstance(obj, User):
+            return self._get_role(obj=obj)
+        return 'base'
+
+    @extend_schema_field(PermissionsSerializer)
+    def get_permissions(self, obj: Union[User, TigaUser]):
+        permissions = []
+        if isinstance(obj, User):
+            permissions = UserStat.get_permissions(
+                role=self.get_role(obj=obj)
+            )
+        elif isinstance(obj, TigaUser):
+            permissions = TigaUser.get_permissions()
+        return PermissionsSerializer(permissions).data
+
+class UserPermissionSerializer(serializers.Serializer):
+    class GeneralPermissionSerializer(BaseRolePermissionSerializer):
+        is_staff = serializers.BooleanField()
+
+        def _get_role(self, obj: User):
+            userstat = UserStat.objects.filter(user=obj).first()
+            if userstat:
+                return userstat.get_role()
+            return 'base'
+
+    class CountryPermissionSerializer(BaseRolePermissionSerializer):
+        def __init__(self, *args, **kwargs):
+            self.country = kwargs.pop('country', None)
+            super().__init__(*args, **kwargs)
+
+        country = serializers.SerializerMethodField()
+
+        def _get_role(self, obj: User):
+            userstat = UserStat.objects.filter(user=obj).first()
+            if userstat:
+                return userstat.get_role(country=self.country)
+            return 'base'
+
+        @extend_schema_field(CountrySerializer)
+        def get_country(self, obj: Union[User, TigaUser]):
+            return CountrySerializer(self.country).data
+
+    general = serializers.SerializerMethodField()
+    countries = serializers.SerializerMethodField()
+
+    @extend_schema_field(GeneralPermissionSerializer)
+    def get_general(self, obj: Union[User, TigaUser]):
+        return self.GeneralPermissionSerializer(obj).data
+
+    @extend_schema_field(CountryPermissionSerializer(many=True))
+    def get_countries(self, obj: Union[User, TigaUser]):
+        if not isinstance(obj, User):
+            return self.CountryPermissionSerializer(many=True).data
+        userstat = UserStat.objects.filter(user=obj).first()
+        if userstat:
+            result = []
+            for country in [userstat.national_supervisor_of, userstat.native_of]:
+                if not country:
+                    continue
+                result.append(self.CountryPermissionSerializer(instance=obj, country=country).data)
+            return result
+        return self.CountryPermissionSerializer(many=True).data
+
 class UserSerializer(serializers.ModelSerializer):
     class UserScoreSerializer(serializers.ModelSerializer):
         value = serializers.IntegerField(source="score_v2", min_value=0, read_only=True)

api/tests/integration/permissions/get_mine.tavern.yml

@@ -0,0 +1,108 @@
+---
+
+test_name: Anonymous user can not get permissions
+
+includes:
+  - !include schema.yml
+
+marks:
+  - usefixtures:
+    - api_live_url
+
+stages:
+  - name: Anonymous user can not get permissions
+    request:
+      url: "{api_live_url}/me/permissions/"
+      method: GET
+    response:
+      status_code: 401
+
+---
+
+test_name: Mobile users get permissions
+
+includes:
+  - !include schema.yml
+
+marks:
+  - usefixtures:
+    - api_live_url
+    - app_user
+    - app_user_token
+
+stages:
+  - name: Mobile users get permissions.
+    request:
+      url: "{api_live_url}/me/permissions/"
+      method: GET
+      headers:
+        Authorization: 'Bearer {app_user_token}'
+    response:
+      status_code: 200
+      json:
+        general:
+          role: "base"
+          permissions:
+            annotation:
+              add: false
+              change: false
+              view: false
+              delete: false
+              mark_as_decisive: false
+            identification_task:
+              add: false
+              change: false
+              view: false
+              delete: false
+            review:
+              add: false
+              change: false
+              view: false
+              delete: false
+          is_staff: false
+        countries: []
+
+---
+
+test_name: Regular users get permissions
+
+includes:
+  - !include schema.yml
+
+marks:
+  - usefixtures:
+    - api_live_url
+    - user
+    - jwt_token_user
+
+stages:
+  - name: Regular users get permissions.
+    request:
+      url: "{api_live_url}/me/permissions/"
+      method: GET
+      headers:
+        Authorization: 'Bearer {jwt_token_user}'
+    response:
+      status_code: 200
+      json:
+        general:
+          role: "base"
+          permissions:
+            annotation:
+              add: false
+              change: false
+              view: false
+              delete: false
+              mark_as_decisive: false
+            identification_task:
+              add: false
+              change: false
+              view: false
+              delete: false
+            review:
+              add: false
+              change: false
+              view: false
+              delete: false
+          is_staff: false
+        countries: []

api/tests/integration/permissions/schema.yml

@@ -0,0 +1,36 @@
+---
+
+name: Common test information
+description: Login information for test server
+
+variables:
+  endpoint: "permissions"
+  response_data_validation:
+    general:
+      role: !anystr
+      permissions: &retrieve_validation
+        annotation:
+          add: !anybool
+          change: !anybool
+          view: !anybool
+          delete: !anybool
+          mark_as_decisive: !anybool
+        identification_task:
+          add: !anybool
+          change: !anybool
+          view: !anybool
+          delete: !anybool
+        review:
+          add: !anybool
+          change: !anybool
+          view: !anybool
+          delete: !anybool
+      is_staff: !anybool
+    countries:
+      country:
+        id: !anyint
+        name_en: !anystr
+        iso3_code: !anystr
+      role: !anystr
+      permissions:
+        <<: *retrieve_validation

api/tests/test_views.py

@@ -6,6 +6,7 @@
 import pytest
 
 from django.contrib.auth import get_user_model
+from django.contrib.auth.models import Group
 from django.utils import timezone
 from django.utils.module_loading import import_string
 
@@ -973,4 +974,119 @@ def test_classification_null_shoud_set_status_to_hidden(self, api_client, endpoi
 
         annotation = ExpertReportAnnotation.objects.get(pk=response.data['id'])
         assert annotation.status == ExpertReportAnnotation.STATUS_HIDDEN
-        assert annotation.taxon is None
+        assert annotation.taxon is None
+
+@pytest.mark.django_db
+class TestPermissionsApi:
+    @pytest.fixture
+    def me_endpoint(self):
+        return f'/api/v1/me/permissions/'
+
+    @pytest.fixture
+    def api_client(self, user):
+        api_client = APIClient()
+        api_client.force_login(user=user)
+
+        return api_client
+
+    def test_general_role_base(self, api_client, me_endpoint):
+        response = api_client.get(
+            me_endpoint,
+            format='json'
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data['general']['role'] == 'base'
+
+    def test_general_role_annotator(self, api_client, user, me_endpoint):
+        expert_group, _ = Group.objects.get_or_create(name='expert')
+        user.groups.add(expert_group)
+
+        response = api_client.get(
+            me_endpoint,
+            format='json'
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data['general']['role'] == 'annotator'
+        assert not response.data['general']['permissions']['review']['add']
+        assert not response.data['general']['permissions']['review']['view']
+
+    def test_general_role_reviewer(self, api_client, user, me_endpoint):
+        superexpert_group, _ = Group.objects.get_or_create(name='superexpert')
+        user.groups.add(superexpert_group)
+
+        response = api_client.get(
+            me_endpoint,
+            format='json'
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data['general']['role'] == 'reviewer'
+        assert response.data['general']['permissions']['review']['add']
+        assert response.data['general']['permissions']['review']['view']
+
+    def test_general_role_admin(self, api_client, user, me_endpoint):
+        user.is_superuser = True
+        user.save()
+
+        response = api_client.get(
+            me_endpoint,
+            format='json'
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data['general']['role'] == 'admin'
+
+    @pytest.mark.parametrize(
+        "is_staff",
+        [True, False]
+    )
+    def test_general_is_staff(self, user, api_client, me_endpoint, is_staff):
+        user.is_staff = is_staff
+        user.save()
+
+        response = api_client.get(
+            me_endpoint,
+            format='json'
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data['general']['is_staff'] == is_staff
+
+    def test_countries_role_supervisor(self, api_client, user, me_endpoint, country):
+        expert_group, _ = Group.objects.get_or_create(name='expert')
+        user.groups.add(expert_group)
+
+        userstat = user.userstat
+        userstat.national_supervisor_of = country
+        userstat.save()
+
+        response = api_client.get(
+            me_endpoint,
+            format='json'
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data['general']['role'] == 'annotator'
+        assert response.data['countries'][0]['country']['id'] == country.pk
+        assert response.data['countries'][0]['role'] == 'supervisor'
+        assert response.data['countries'][0]['permissions']['annotation']['mark_as_decisive']
+        assert response.data['countries'][0]['permissions']['identification_task']['view']
+        assert not response.data['countries'][0]['permissions']['review']['add']
+        assert not response.data['countries'][0]['permissions']['review']['view']
+
+    def test_countries_role_annotator(self, api_client, user, me_endpoint, country):
+        expert_group, _ = Group.objects.get_or_create(name='expert')
+        user.groups.add(expert_group)
+
+        userstat = user.userstat
+        userstat.native_of = country
+        userstat.save()
+
+        response = api_client.get(
+            me_endpoint,
+            format='json'
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data['general']['role'] == 'annotator'
+        assert response.data['countries'][0]['role'] == 'annotator'
+        assert response.data['countries'][0]['country']['id'] == country.pk
+        assert not response.data['countries'][0]['permissions']['annotation']['mark_as_decisive']
+        assert not response.data['countries'][0]['permissions']['identification_task']['view']
+        assert not response.data['countries'][0]['permissions']['review']['add']
+        assert not response.data['countries'][0]['permissions']['review']['view']