Merge branch 'develop' into feature/1515-setup-sentry-for-mobile

backend/api/v1/v1_forms/views.py

@@ -197,9 +197,8 @@ def get_queryset(self):
             )
         else:
             allowed_path = self.request.user.user_access.administration.path
-            allowed_path += (
-                f"{user_administration_id}" if user_administration_id else ""
-            )
+            if user_administration_id:
+                allowed_path += f"{user_administration_id}"
             queryset = (
                 FormCertificationAssignment.objects.prefetch_related(
                     "administrations"

backend/api/v1/v1_jobs/administrations_bulk_upload.py

@@ -3,6 +3,7 @@
 from django.db import transaction
 from django.db.models import Q
 import pandas as pd
+import numpy as np
 from typing import Any, Dict, List, Tuple, Type, Union
 from django.db.models import Model
 from pandas.core.frame import itertools
@@ -21,17 +22,30 @@
 def seed_administration_data(io_file):
     df = pd.read_excel(io_file, sheet_name='data')
     columns = list(df)
+    columns = [col for col in columns if 'Code' not in col]
     level_count = Levels.objects.count()
     level_map = map_column_model(columns[:level_count], Levels)
     attribute_map = map_column_model(
-            columns[level_count:], AdministrationAttribute)
+        columns[level_count:],
+        AdministrationAttribute
+    )
     records = df.to_dict('records')
     for row in records:
         administration_data = []
         for col, level in level_map.items():
             if bool(pd.isnull(row[col])):
                 break
-            administration_data.append((level, row[col]))
+            administration_code = None
+            if (
+                f"{col} Code" in row and
+                not np.isnan(row[f"{col} Code"])
+            ):
+                administration_code = row[f"{col} Code"]
+            administration_data.append((
+                level,
+                row[col],
+                administration_code
+            ))
         target_administration = seed_administrations(administration_data)
         if not target_administration:
             break
@@ -44,18 +58,21 @@ def seed_administration_data(io_file):
 
 
 def seed_administrations(
-        data: List[Tuple[Levels, str]]) -> Union[Administration, None]:
+        data: List[Tuple[Levels, str, str]]
+        ) -> Union[Administration, None]:
     last_obj = None
     for item in data:
-        level, name = item
+        level, name, code = item
         obj = Administration.objects.filter(
             Q(name__iexact=name),
+            code=code,
             level=level,
             parent=last_obj,
         ).first()
         if not obj:
             obj = Administration.objects.create(
                 name=name.title(),
+                code=code,
                 level=level,
                 parent=last_obj
             )
@@ -121,6 +138,7 @@ def validate_administrations_bulk_upload(io_file):
             "error_message": ValidationText.file_empty_validation.value,
         }]
     headers = list(df)
+    headers = [h for h in headers if 'Code' not in h]
     header_count = len(headers)
     levels = list(Levels.objects.all())
     level_count = len(levels)

backend/api/v1/v1_jobs/tests/test_administrations_bulk_upload.py

@@ -78,19 +78,29 @@ def test_seed_data(self):
             self.option_att.id])
         df.loc[len(df)] = [
                 'Indonesia',
+                '001',
                 'Jakarta',
+                '101',
                 'South Jakarta',
+                '201',
                 'Pasar Minggu',
+                '301',
                 'Lenteng Agung',
+                '401',
                 'y',
                 'opt #1'
                 ]
         df.loc[len(df)] = [
                 'Indonesia',
+                '001',
                 'Jakarta',
+                '101',
                 'South Jakarta',
+                '201',
                 'Pasar Minggu',
+                '301',
                 'Jagakarsa',
+                '402',
                 '',
                 'opt #2'
                 ]
@@ -117,10 +127,15 @@ def test_seed_attribute_type_variant(self):
             self.aggregate_att.id])
         df.loc[len(df)] = [
                 'Indonesia',
+                '001',
                 'Jakarta',
+                '101',
                 'South Jakarta',
+                '201',
                 'Pasar Minggu',
+                '301',
                 'Lenteng Agung',
+                '401',
                 '1',
                 'opt #1',
                 'opt #1 | opt #2',
@@ -156,19 +171,29 @@ def test_set_attributes_on_upper_level(self):
             self.option_att.id])
         df.loc[len(df)] = [
                 'Indonesia',
+                '001',
                 'Jakarta',
+                '101',
                 'South Jakarta',
+                '201',
                 'Pasar Minggu',
+                '301',
                 'Lenteng Agung',
+                '401',
                 'y',
                 'opt #1'
                 ]
         df.loc[len(df)] = [
                 'Indonesia',
+                '001',
                 'Jakarta',
+                '101',
                 'South Jakarta',
+                '201',
                 '',
+                None,
                 '',
+                None,
                 'y',
                 ''
                 ]
@@ -181,6 +206,45 @@ def test_set_attributes_on_upper_level(self):
         level3 = Administration.objects.get(name='South Jakarta')
         self.assertEqual(1, level3.attributes.count())
 
+    def test_update_existing_attribute_values(self):
+        df = generate_inmemory_template([
+            self.value_att.id,
+            self.option_att.id,
+            self.aggregate_att.id])
+        df.loc[len(df)] = [
+            'Indonesia',
+            '001',
+            'Jakarta',
+            '101',
+            'South Jakarta',
+            '201',
+            'Pasar Minggu',
+            '301',
+            'Lenteng Agung',
+            '401',
+            '999',
+            'opt #2',
+            '47',
+            '53',
+        ]
+        upload_file = write_inmemory_excel_file(df)
+        seed_administration_data(upload_file)
+
+        adm1 = Administration.objects.get(name='Lenteng Agung')
+        self.assertEqual(3, adm1.attributes.count())
+
+        value_att = adm1.attributes.get(attribute=self.value_att)
+        self.assertEqual('999', value_att.value.get('value'))
+
+        option_att = adm1.attributes.get(attribute=self.option_att)
+        self.assertEqual('opt #2', option_att.value.get('value'))
+
+        aggregate_att = adm1.attributes.get(attribute=self.aggregate_att)
+        self.assertEqual(
+            {'opt #1': '47', 'opt #2': '53'},
+            aggregate_att.value.get('value')
+        )
+
 
 class ValidateBulkUploadTestCase(TestCase, ProfileTestHelperMixin):
     def setUp(self):
@@ -324,10 +388,15 @@ def test_invalid_attribute_values(self):
             self.aggregate_att.id])
         df.loc[len(df)] = [
                 'Indonesia',
+                '001',
                 'Jakarta',
+                '101',
                 'South Jakarta',
+                '201',
                 'Pasar Minggu',
+                '301',
                 'Lenteng Agung',
+                '401',
                 '1',
                 'invalid',
                 'opt #1 | invalid',

backend/api/v1/v1_jobs/tests/tests_job_download_command.py

@@ -8,6 +8,7 @@
 )
 from api.v1.v1_users.models import SystemUser
 from api.v1.v1_profile.constants import UserRoleTypes
+from api.v1.v1_profile.models import Administration
 
 
 @override_settings(USE_TZ=False)
@@ -44,3 +45,31 @@ def test_download_all_data(self):
 
         url = job_generate_data_download(job_id=job.id, **job.info)
         self.assertTrue("download-test_form" in url)
+
+    def test_download_recent_data_with_administration(self):
+        form = Forms.objects.get(pk=1)
+        admin = SystemUser.objects.filter(
+            user_access__role=UserRoleTypes.admin
+        ).first()
+
+        ward = Administration.objects.filter(
+            level__name="Ward"
+        ).order_by("?").first()
+
+        result = call_command(
+            "job_download",
+            form.id,
+            admin.id,
+            "-a",
+            ward.id,
+            "-t",
+            "recent",
+        )
+        self.assertTrue(result)
+
+        job = Jobs.objects.get(pk=result)
+        self.assertEqual(job.info.get("download_type"), "recent")
+        self.assertEqual(job.info.get("administration"), ward.id)
+
+        url = job_generate_data_download(job_id=job.id, **job.info)
+        self.assertTrue("download-test_form" in url)

backend/api/v1/v1_mobile/tests/tests_subordinates_mobile_users.py

@@ -35,8 +35,9 @@ def setUp(self):
             administration=administration.parent
         )
         self.token = self.get_auth_token(self.user.email)
-        form = Forms.objects.first()
+        form = Forms.objects.get(pk=1)
         self.form = form
+        self.form2 = Forms.objects.get(pk=2)
         UserForms.objects.get_or_create(form=form, user=self.user)
         UserForms.objects.get_or_create(form=form, user=self.approver)
 
@@ -105,3 +106,39 @@ def test_same_level_mobile_users_list(self):
         body = response.json()
         self.assertEqual(len(body['data']), 1)
         self.assertTrue(True)
+
+
+def test_subordinates_with_diff_forms(self):
+    payload = {
+        'name': 'user3 assignment',
+        'forms': [self.form2.id],
+        'administrations': [self.user.user_access.administration.id],
+    }
+
+    response = typing.cast(
+        HttpResponse,
+        self.client.post(
+            '/api/v1/mobile-assignments',
+            payload,
+            content_type="application/json",
+            HTTP_AUTHORIZATION=f'Bearer {self.token}'
+        )
+    )
+
+    self.assertEqual(response.status_code, 201)
+
+    # Login as Sub-county user
+    t = RefreshToken.for_user(self.approver)
+
+    response = typing.cast(
+        HttpResponse,
+        self.client.get(
+            '/api/v1/mobile-assignments',
+            content_type="application/json",
+            HTTP_AUTHORIZATION=f'Bearer {t.access_token}'
+        )
+    )
+    self.assertEqual(response.status_code, 200)
+    body = response.json()
+    # Get all sub-ordinate mobile users
+    self.assertEqual(len(body['data']), 2)

backend/api/v1/v1_mobile/views.py

@@ -393,35 +393,34 @@ class MobileAssignmentViewSet(ModelViewSet):
 
     def get_queryset(self):
         user = self.request.user
-        adm_path = user.user_access.administration.path
-        adm_path += f"{user.user_access.administration.id}"
         mobile_users = MobileAssignment.objects.prefetch_related(
             "administrations", "forms"
         ).filter(user=user)
 
-        user_form_subquery = UserForms.objects.filter(
-            user=user,
-            form_id=OuterRef("forms__pk"),
-        ).values("form_id")
-
-        exclude_ids = (
-            MobileAssignment.objects.filter(
-                administrations__path__startswith=adm_path
-            )
-            .annotate(has_matching_user_form=Exists(user_form_subquery))
-            .filter(has_matching_user_form=False)
-            .distinct()
-        )
-
-        mobile_users |= (
-            MobileAssignment.objects.prefetch_related(
+        user_adm = user.user_access.administration
+        adm_path = f"{user_adm.path}{user_adm.id}"
+        descendant_users = MobileAssignment.objects.prefetch_related(
                 "administrations", "forms"
-            )
+            ) \
             .filter(
                 administrations__path__startswith=adm_path,
             )
-            .exclude(pk__in=exclude_ids)
-        )
+        if user_adm.level.level > 2:
+            # Check if the user is under Sub-County
+            user_form_subquery = UserForms.objects.filter(
+                user=user,
+                form_id=OuterRef("forms__pk"),
+            ).values("form_id")
+            exclude_ids = (
+                MobileAssignment.objects.filter(
+                    administrations__path__startswith=adm_path
+                )
+                .annotate(has_matching_user_form=Exists(user_form_subquery))
+                .filter(has_matching_user_form=False)
+                .distinct()
+            )
+            descendant_users = descendant_users.exclude(pk__in=exclude_ids)
+        mobile_users |= descendant_users
         return mobile_users.order_by("-id").distinct()
 
 

backend/api/v1/v1_profile/tests/test_export_template.py

@@ -50,8 +50,9 @@ def test_export_template(self):
         self.assertEqual(response["Content-Type"], self.XLSX_MIME)
         df = pd.read_excel(response.content, sheet_name="data")
         expected = [
-            f"{lvl.id}|{lvl.name}"
-            for lvl in Levels.objects.order_by("level").all()
+            col
+            for lvl in Levels.objects.order_by('level').all()
+            for col in [f'{lvl.id}|{lvl.name}', f'{lvl.id}|{lvl.name} Code']
         ]
         actual = [val for val in list(df)]
         self.assertEqual(expected, actual)
@@ -69,9 +70,13 @@ def test_with_attribute(self):
             ),
         )
         self.assertEqual(response.status_code, 200)
-        self.assertEqual(response["Content-Type"], self.XLSX_MIME)
-        df = pd.read_excel(response.content, sheet_name="data")
-        levels = [f"{lvl.id}|{lvl.name}" for lvl in Levels.objects.all()]
+        self.assertEqual(response['Content-Type'], self.XLSX_MIME)
+        df = pd.read_excel(response.content, sheet_name='data')
+        levels = [
+            col
+            for lvl in Levels.objects.all()
+            for col in [f'{lvl.id}|{lvl.name}', f'{lvl.id}|{lvl.name} Code']
+        ]
         attributes = [
             f"{self.attribute1.id}|{self.attribute1.name}",
             f"{self.attribute2.id}|{self.attribute2.name}|opt #1",