Merge branch 'develop' into selenium/targeting_smoke_tests

.github/workflows/ci.yml

@@ -240,7 +240,6 @@ jobs:
           token: ${{ secrets.CODECOV_TOKEN }}
           verbose: true
 
-
   trivy:
     runs-on: ubuntu-latest
     needs: [build_and_push_dist]

backend/.coveragerc

@@ -4,9 +4,12 @@ source = hct_mis_api
 concurrency=multiprocessing
 
 omit =
+        */selenium_tests/**
         */tests/**
         */migrations/*,
         */apps.py,
+        */admin/*.py
+        */admin.py
         hct_mis_api/one_time_scripts/*,
         hct_mis_api/libs/*,
         hct_mis_api/settings/*,
@@ -30,8 +33,9 @@ exclude_lines =
     # Don't complain if non-runnable code isn't run:
     #if 0:
     if __name__ == .__main__.:
+    if TYPE_CHECKING
 
-fail_under = 20
+fail_under = 15
 
 ignore_errors = False
 

backend/dev.sh

@@ -19,7 +19,8 @@ else
         --reruns-delay 1 \
         --cov-report xml:coverage.xml \
         --randomly-seed=42 \
-        hct_mis_api/
+        hct_mis_api/ \
+        tests/
       ;;
     "lint")
       mkdir -p ./lint-results

backend/hct_mis_api/api/caches.py

@@ -0,0 +1,55 @@
+import functools
+from typing import Any, Callable
+
+from django.core.cache import cache
+
+from rest_framework import status
+from rest_framework.response import Response
+from rest_framework_extensions.key_constructor.constructors import KeyConstructor
+
+
+def etag_decorator(key_constructor_class: "KeyConstructor", compare_etags: bool = True) -> Callable:
+    """
+    Decorator operating on ViewSet methods.
+    It expects for the first argument to be a view instance and for the second to be a request like:
+        def view_function(self, request, *args, **kwargs)
+    It calculates etag based on the KeyConstructor (key_constructor_class) and adds it to the response.
+    If compare_etags is True it compares calculated etag with If-None_match header from the request.
+    If they are the same, it returns 304 status code without any data.
+    """
+
+    def inner(function: Callable) -> Callable:
+        @functools.wraps(function)
+        def wrapper(*args: dict, **kwargs: dict) -> Response:
+            # in view methods first argument is always self, second is request
+            request = args[1]
+            etag = key_constructor_class()(
+                view_instance=args[0],
+                view_method=function,
+                request=request,
+                args=args[2:],
+                kwargs=kwargs,
+            )
+
+            # If etag from header and calculated are the same,
+            # return 304 status code as request consists of the same data already (cached on client side)
+            if compare_etags and request.headers.get("If-None-Match") == etag:
+                return Response(status=status.HTTP_304_NOT_MODIFIED, headers={"ETAG": etag})
+            res = function(*args, **kwargs)
+            res.headers["ETAG"] = etag
+            return res
+
+        return wrapper
+
+    return inner
+
+
+def get_or_create_cache_key(key: str, default: Any) -> Any:
+    """
+    Get value from cache by key or create it with default value.
+    """
+    value = cache.get(key)
+    if value is None:
+        cache.set(key, default, timeout=None)
+        return default
+    return value

backend/hct_mis_api/apps/account/admin/partner.py

@@ -1,5 +1,5 @@
 from collections import defaultdict
-from typing import Any, Optional, Sequence, Type, Union
+from typing import TYPE_CHECKING, Any, Optional, Sequence, Type, Union
 
 from django import forms
 from django.contrib import admin, messages
@@ -12,21 +12,31 @@
 from admin_extra_buttons.decorators import button
 
 from hct_mis_api.apps.account import models as account_models
-from hct_mis_api.apps.account.models import IncompatibleRoles, PartnerPermission, Role
-from hct_mis_api.apps.core.models import BusinessArea
+from hct_mis_api.apps.account.models import IncompatibleRoles, Role
+from hct_mis_api.apps.core.models import BusinessArea, BusinessAreaPartnerThrough
 from hct_mis_api.apps.geo.models import Area
 from hct_mis_api.apps.program.models import Program
 from hct_mis_api.apps.utils.admin import HopeModelAdminMixin
 from mptt.forms import TreeNodeMultipleChoiceField
 
+if TYPE_CHECKING:
+    from django.db.models.query import QuerySet
+
 
 def can_add_business_area_to_partner(request: Any, *args: Any, **kwargs: Any) -> bool:
     return request.user.can_add_business_area_to_partner()
 
 
-class BusinessAreaRoleForm(forms.Form):
-    business_area = forms.ModelChoiceField(queryset=BusinessArea.objects.all(), required=True)
-    roles = forms.ModelMultipleChoiceField(queryset=Role.objects.all(), required=True)
+def business_area_role_form_custom_query(queryset: "QuerySet") -> Any:
+    class BusinessAreaRoleForm(forms.Form):
+        business_area = forms.ModelChoiceField(queryset=BusinessArea.objects.all(), required=True)
+        roles = forms.ModelMultipleChoiceField(queryset=Role.objects.all(), required=True)
+
+        def __init__(self, *args: Any, **kwargs: Any) -> None:
+            super().__init__(*args, **kwargs)
+            self.fields["business_area"].queryset = queryset
+
+    return BusinessAreaRoleForm
 
 
 class ProgramAreaForm(forms.Form):
@@ -86,98 +96,70 @@ def permissions(self, request: HttpRequest, pk: int) -> Union[TemplateResponse,
         context = self.get_common_context(request, pk, title="Partner permissions")
         partner: account_models.Partner = context["original"]
         user_can_add_ba_to_partner = request.user.can_add_business_area_to_partner()
-        permissions_list = partner.get_permissions().to_list()
+        permissions_list = partner.business_area_partner_through.all()
         context["can_add_business_area_to_partner"] = user_can_add_ba_to_partner
 
-        BusinessAreaRoleFormSet = formset_factory(BusinessAreaRoleForm, extra=0, can_delete=True)
-        ProgramAreaFormSet = formset_factory(ProgramAreaForm, extra=0, can_delete=True)
-
-        business_areas = set()
-
+        BusinessAreaRoleFormSet = formset_factory(
+            business_area_role_form_custom_query(partner.allowed_business_areas.all()),
+            extra=0,
+            can_delete=True,
+        )
         if request.method == "GET":
             business_area_role_data = []
-            program_area_data = []
             for permission in permissions_list:
                 if permission.roles:
                     business_area_role_data.append(
-                        {"business_area": permission.business_area_id, "roles": permission.roles}
+                        {"business_area": permission.business_area_id, "roles": permission.roles.all()}
                     )
-            for permission in permissions_list:
-                for program_id, areas in permission.programs.items():
-                    program_area_data.append(
-                        {"business_area": permission.business_area_id, "program": program_id, "areas": areas}
-                    )
-                    business_areas.add(permission.business_area_id)
             business_area_role_form_set = BusinessAreaRoleFormSet(
                 initial=business_area_role_data, prefix="business_area_role"
             )
-            program_area_form_set = ProgramAreaFormSet(initial=program_area_data, prefix="program_areas")
         else:
-            partner_permissions = PartnerPermission()
             business_area_role_form_set = BusinessAreaRoleFormSet(request.POST or None, prefix="business_area_role")
-            program_area_form_set = ProgramAreaFormSet(request.POST or None, prefix="program_areas")
-            refresh_areas = request.POST["refresh-areas"]
             incompatible_roles = defaultdict(list)
+            ba_partner_through_data = {}
+            ba_partner_through_to_be_deleted = []
 
             business_area_role_form_set_is_valid = business_area_role_form_set.is_valid()
             if user_can_add_ba_to_partner and business_area_role_form_set_is_valid:
                 for form in business_area_role_form_set.cleaned_data:
                     if form and not form["DELETE"]:
                         business_area_id = str(form["business_area"].id)
                         role_ids = list(map(lambda role: str(role.id), form["roles"]))
-                        partner_permissions.set_roles(business_area_id, role_ids)
 
                         if incompatible_role := IncompatibleRoles.objects.filter(
                             role_one__in=role_ids, role_two__in=role_ids
                         ).first():
                             incompatible_roles[form["business_area"]].append(str(incompatible_role))
                         else:
-                            partner_permissions.set_roles(business_area_id, role_ids)
-            # save the same BA and roles for user without perm
-            if not user_can_add_ba_to_partner:
-                business_area_role_form_set_is_valid = True
-                for permission in permissions_list:
-                    if permission.roles:
-                        partner_permissions.set_roles(permission.business_area_id, permission.roles)
-
-            if program_area_form_set.is_valid():
-                for form in program_area_form_set.cleaned_data:
-                    if form and not form["DELETE"]:
-                        business_area_id = str(form["business_area"].id)
-                        program_id = str(form["program"].id)
-                        areas_ids = list(map(lambda area: str(area.id), form["areas"]))
-                        partner_permissions.set_program_areas(business_area_id, program_id, areas_ids)
-
-            for program_area_form in program_area_form_set:
-                if program_area_form.cleaned_data.get("business_area"):
-                    business_areas.add(program_area_form.cleaned_data["business_area"].id)
+                            ba_partner, _ = BusinessAreaPartnerThrough.objects.get_or_create(
+                                partner=partner,
+                                business_area_id=business_area_id,
+                            )
+                            ba_partner_through_data[ba_partner] = form["roles"]
+                    elif form["DELETE"]:
+                        ba_partner_through_to_be_deleted.append(
+                            BusinessAreaPartnerThrough.objects.filter(
+                                partner=partner, business_area=form["business_area"]
+                            )
+                            .first()
+                            .id
+                        )
 
             if incompatible_roles:
                 for business_area, roles in incompatible_roles.items():
                     self.message_user(
-                        request, (f"Roles in {business_area} are incompatible: {', '.join(roles)}"), messages.ERROR
+                        request, f"Roles in {business_area} are incompatible: {', '.join(roles)}", messages.ERROR
                     )
 
-            if (
-                refresh_areas == "false"
-                and business_area_role_form_set_is_valid
-                and program_area_form_set.is_valid()
-                and not incompatible_roles
-            ):
-                partner.set_permissions(partner_permissions)
-                partner.save()
+            if business_area_role_form_set_is_valid and not incompatible_roles:
+                if ba_partner_through_to_be_deleted:
+                    BusinessAreaPartnerThrough.objects.filter(pk__in=ba_partner_through_to_be_deleted).delete()
+                for ba_partner_through, areas in ba_partner_through_data.items():
+                    ba_partner_through.roles.add(*areas)
 
                 return HttpResponseRedirect(reverse("admin:account_partner_change", args=[pk]))
 
         context["business_area_role_formset"] = business_area_role_form_set
-        context["program_area_formset"] = program_area_form_set
-        context["areas"] = {}
-        context["program"] = {}
-
-        for business_area_id in business_areas:
-            context["areas"][str(business_area_id)] = Area.objects.filter(
-                area_type__country__business_areas__id=business_area_id
-            )
-            context["program"][str(business_area_id)] = Program.objects.filter(business_area_id=business_area_id)
 
         return TemplateResponse(request, "admin/account/parent/permissions.html", context)

backend/hct_mis_api/apps/account/filters.py

@@ -7,7 +7,6 @@
 from django_filters import BooleanFilter, CharFilter, FilterSet, MultipleChoiceFilter
 
 from hct_mis_api.apps.account.models import USER_STATUS_CHOICES, Partner, Role
-from hct_mis_api.apps.core.models import BusinessArea
 from hct_mis_api.apps.core.utils import CustomOrderingFilter
 
 if TYPE_CHECKING:
@@ -66,11 +65,9 @@ def search_filter(self, qs: "QuerySet", name: str, value: str) -> "QuerySet[User
         return qs.filter(q_obj)
 
     def business_area_filter(self, qs: "QuerySet", name: str, value: str) -> "QuerySet[User]":
-        business_area_id = BusinessArea.objects.get(slug=value).id
         return qs.filter(
             Q(user_roles__business_area__slug=value)
-            | Q(partner__permissions__has_key=str(business_area_id))
-            | Q(partner__name="UNICEF")
+            | Q(partner__business_area_partner_through__business_area__slug=value)
         )
 
     def partners_filter(self, qs: "QuerySet", name: str, values: List["UUID"]) -> "QuerySet[User]":