Merge branch 'develop' into selenium/feedback

backend/hct_mis_api/api/endpoints/rdi/__init__.py

@@ -4,7 +4,11 @@
     PushLaxToRDIView,
     PushToRDIView,
 )
+from hct_mis_api.api.endpoints.rdi.delegate_people import (  # noqa: F401
+    DelegatePeopleRDIView,
+)
 from hct_mis_api.api.endpoints.rdi.program import ProgramViewSet  # noqa: F401
+from hct_mis_api.api.endpoints.rdi.push_people import PushPeopleToRDIView  # noqa: F401
 from hct_mis_api.api.endpoints.rdi.upload import UploadRDIView  # noqa: F401
 from hct_mis_api.api.endpoints.rdi.upload_people import (  # noqa: F401
     UploadPeopleRDIView,

backend/hct_mis_api/api/endpoints/rdi/delegate_people.py

@@ -0,0 +1,52 @@
+from typing import Dict
+from uuid import UUID
+
+from django.db.transaction import atomic
+
+from drf_yasg.utils import swagger_auto_schema
+from rest_framework import serializers, status
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from hct_mis_api.api.endpoints.base import HOPEAPIBusinessAreaView, HOPEAPIView
+from hct_mis_api.api.models import Grant
+from hct_mis_api.apps.household.models import ROLE_PRIMARY
+from hct_mis_api.apps.registration_datahub.models import (
+    ImportedIndividualRoleInHousehold,
+)
+
+
+class DelegateSerializer(serializers.Serializer):
+    delegate_id = serializers.UUIDField(required=True)
+    delegated_for = serializers.ListField(child=serializers.UUIDField(required=True), allow_empty=False, required=True)
+
+
+class DelegatePeopleSerializer(serializers.Serializer):
+    delegates = DelegateSerializer(many=True, required=True, allow_empty=False, allow_null=False)
+
+    def create(self, validated_data: Dict) -> Dict:
+        delegates = validated_data.pop("delegates")
+        updated = 0
+        for delegate in delegates:
+            delegate_id = delegate["delegate_id"]
+            delegated_for = delegate["delegated_for"]
+            for delegated_for_id in delegated_for:
+                updated += ImportedIndividualRoleInHousehold.objects.filter(
+                    household__individuals__in=[delegated_for_id],
+                    individual_id=delegated_for_id,
+                    role=ROLE_PRIMARY,
+                ).update(individual_id=delegate_id)
+        return {"updated": updated}
+
+
+class DelegatePeopleRDIView(HOPEAPIBusinessAreaView, HOPEAPIView):
+    permission = Grant.API_RDI_UPLOAD
+
+    @swagger_auto_schema(request_body=DelegatePeopleSerializer)
+    @atomic(using="registration_datahub")
+    def post(self, request: "Request", business_area: str, rdi: UUID) -> Response:
+        serializer = DelegatePeopleSerializer(data=request.data)
+        if serializer.is_valid():
+            response = serializer.save()
+            return Response(response, status=status.HTTP_200_OK)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

backend/hct_mis_api/api/endpoints/rdi/push_people.py

@@ -0,0 +1,169 @@
+from typing import Dict, List, Optional
+from uuid import UUID
+
+from django.db.transaction import atomic
+from django.http import Http404
+from django.utils import timezone
+from django.utils.functional import cached_property
+
+from drf_yasg.utils import swagger_auto_schema
+from rest_framework import serializers, status
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from hct_mis_api.api.endpoints.base import HOPEAPIBusinessAreaView, HOPEAPIView
+from hct_mis_api.api.endpoints.rdi.mixin import get_photo_from_stream
+from hct_mis_api.api.endpoints.rdi.upload import BirthDateValidator, DocumentSerializer
+from hct_mis_api.api.models import Grant
+from hct_mis_api.apps.household.models import (
+    BLANK,
+    COLLECT_TYPES,
+    HEAD,
+    NON_BENEFICIARY,
+    RESIDENCE_STATUS_CHOICE,
+    ROLE_PRIMARY,
+)
+from hct_mis_api.apps.registration_data.models import RegistrationDataImport
+from hct_mis_api.apps.registration_datahub.models import (
+    ImportedDocument,
+    ImportedDocumentType,
+    ImportedHousehold,
+    ImportedIndividual,
+    RegistrationDataImportDatahub,
+)
+
+PEOPLE_TYPE_CHOICES = (
+    (BLANK, "None"),
+    (NON_BENEFICIARY, "Non Beneficiary"),
+)
+
+
+class PeopleSerializer(serializers.ModelSerializer):
+    first_registration_date = serializers.DateTimeField(default=timezone.now)
+    last_registration_date = serializers.DateTimeField(default=timezone.now)
+    observed_disability = serializers.CharField(allow_blank=True, required=False)
+    marital_status = serializers.CharField(allow_blank=True, required=False)
+    documents = DocumentSerializer(many=True, required=False)
+    birth_date = serializers.DateField(validators=[BirthDateValidator()])
+
+    type = serializers.ChoiceField(choices=PEOPLE_TYPE_CHOICES, required=True)
+
+    country_origin = serializers.CharField(allow_blank=True, required=False)
+    country = serializers.CharField(allow_blank=True, required=True)
+    collect_individual_data = serializers.ChoiceField(choices=COLLECT_TYPES)
+    residence_status = serializers.ChoiceField(choices=RESIDENCE_STATUS_CHOICE)
+    village = serializers.CharField(allow_blank=True, required=False)
+
+    class Meta:
+        model = ImportedIndividual
+        exclude = [
+            "id",
+            "registration_data_import",
+            "deduplication_batch_results",
+            "deduplication_golden_record_results",
+            "deduplication_batch_status",
+            "created_at",
+            "updated_at",
+            "mis_unicef_id",
+            "household",
+            "kobo_asset_id",
+            "row_id",
+            "detail_id",
+        ]
+
+
+class PeopleUploadMixin:
+    def save_people(self, rdi: RegistrationDataImportDatahub, program_id: UUID, people_data: List[Dict]) -> List[int]:
+        people_ids = []
+        for person_data in people_data:
+            documents = person_data.pop("documents", [])
+
+            hh = self._create_household(person_data, program_id, rdi)
+            ind = self._create_individual(documents, hh, person_data, program_id, rdi)
+            people_ids.append(ind.id)
+        return people_ids
+
+    def _create_household(
+        self, person_data: Dict, program_id: UUID, rdi: RegistrationDataImportDatahub
+    ) -> Optional[ImportedHousehold]:
+        if person_data.get("type") == NON_BENEFICIARY:
+            return None
+        household_fields = [field.name for field in ImportedHousehold._meta.get_fields()]
+        household_data = {field: value for field, value in person_data.items() if field in household_fields}
+        return ImportedHousehold.objects.create(
+            registration_data_import=rdi,
+            program_id=program_id,
+            collect_type=ImportedHousehold.CollectType.SINGLE.value,
+            **household_data,
+        )
+
+    def _create_individual(
+        self,
+        documents: List[Dict],
+        hh: Optional[ImportedHousehold],
+        person_data: Dict,
+        program_id: UUID,
+        rdi: RegistrationDataImportDatahub,
+    ) -> ImportedIndividual:
+        individual_fields = [field.name for field in ImportedIndividual._meta.get_fields()]
+        individual_data = {field: value for field, value in person_data.items() if field in individual_fields}
+        person_type = person_data.get("type")
+        relationship = NON_BENEFICIARY if person_type is NON_BENEFICIARY else HEAD
+
+        ind = ImportedIndividual.objects.create(
+            household=hh,
+            registration_data_import=rdi,
+            program_id=program_id,
+            relationship=relationship,
+            **individual_data,
+        )
+
+        if person_type is not NON_BENEFICIARY:
+            hh.head_of_household = ind
+            hh.individuals_and_roles.create(individual=ind, role=ROLE_PRIMARY)
+            hh.save()
+
+        for doc in documents:
+            self._create_document(ind, doc)
+        return ind
+
+    def _create_document(self, member: ImportedIndividual, doc: Dict) -> None:
+        ImportedDocument.objects.create(
+            document_number=doc["document_number"],
+            photo=get_photo_from_stream(doc.get("image", None)),
+            doc_date=doc["doc_date"],
+            individual=member,
+            country=doc["country"],
+            type=ImportedDocumentType.objects.get(key=doc["type"]),
+        )
+
+
+class PushPeopleToRDIView(HOPEAPIBusinessAreaView, PeopleUploadMixin, HOPEAPIView):
+    permission = Grant.API_RDI_UPLOAD
+
+    @cached_property
+    def selected_rdi(self) -> RegistrationDataImportDatahub:
+        try:
+            return RegistrationDataImportDatahub.objects.get(
+                import_done=RegistrationDataImportDatahub.LOADING,
+                id=self.kwargs["rdi"],
+                business_area_slug=self.kwargs["business_area"],
+            )
+        except RegistrationDataImportDatahub.DoesNotExist:
+            raise Http404
+
+    @swagger_auto_schema(request_body=PeopleSerializer)
+    @atomic(using="registration_datahub")
+    def post(self, request: "Request", business_area: str, rdi: UUID) -> Response:
+        serializer = PeopleSerializer(data=request.data, many=True)
+        program_id = RegistrationDataImport.objects.get(datahub_id=str(self.selected_rdi.id)).program_id
+
+        if serializer.is_valid():
+            people_ids = self.save_people(self.selected_rdi, program_id, serializer.validated_data)
+
+            response = {
+                "id": self.selected_rdi.id,
+                "people": people_ids,
+            }
+            return Response(response, status=status.HTTP_201_CREATED)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

backend/hct_mis_api/api/tests/test_delegate_people.py

@@ -0,0 +1,127 @@
+from typing import List
+from uuid import UUID
+
+from rest_framework import status
+from rest_framework.reverse import reverse
+
+from hct_mis_api.api.models import Grant
+from hct_mis_api.api.tests.base import HOPEApiTestCase
+from hct_mis_api.apps.core.fixtures import DataCollectingTypeFactory
+from hct_mis_api.apps.core.models import DataCollectingType
+from hct_mis_api.apps.core.utils import IDENTIFICATION_TYPE_TO_KEY_MAPPING
+from hct_mis_api.apps.household.models import (
+    COLLECT_TYPE_FULL,
+    IDENTIFICATION_TYPE_BIRTH_CERTIFICATE,
+    NON_BENEFICIARY,
+)
+from hct_mis_api.apps.program.fixtures import ProgramFactory
+from hct_mis_api.apps.program.models import Program
+from hct_mis_api.apps.registration_data.models import RegistrationDataImport
+from hct_mis_api.apps.registration_datahub.models import (
+    ImportedDocumentType,
+    ImportedHousehold,
+    ImportedIndividual,
+    RegistrationDataImportDatahub,
+)
+
+
+class TestDelegatePeople(HOPEApiTestCase):
+    user_permissions = [Grant.API_RDI_UPLOAD]
+
+    @classmethod
+    def setUpTestData(cls) -> None:
+        super().setUpTestData()
+        ImportedDocumentType.objects.create(
+            key=IDENTIFICATION_TYPE_TO_KEY_MAPPING[IDENTIFICATION_TYPE_BIRTH_CERTIFICATE], label="--"
+        )
+        data_collecting_type = DataCollectingTypeFactory(
+            label="Full",
+            code="full",
+            weight=1,
+            business_areas=[cls.business_area],
+            type=DataCollectingType.Type.SOCIAL.value,
+        )
+        cls.program = ProgramFactory.create(
+            status=Program.DRAFT, business_area=cls.business_area, data_collecting_type=data_collecting_type
+        )
+        cls.rdi = RegistrationDataImportDatahub.objects.create(
+            business_area_slug=cls.business_area.slug, import_done=RegistrationDataImportDatahub.LOADING
+        )
+        cls.rdi2: RegistrationDataImport = RegistrationDataImport.objects.create(
+            business_area=cls.business_area,
+            number_of_individuals=0,
+            number_of_households=0,
+            datahub_id=cls.rdi.pk,
+            status=RegistrationDataImport.LOADING,
+            program=cls.program,
+        )
+
+        cls.url = reverse("api:rdi-delegate-people", args=[cls.business_area.slug, str(cls.rdi.id)])
+
+    def test_external_collector(self) -> None:
+        people_ids = self._create_people()
+        households_count = ImportedHousehold.objects.filter(registration_data_import=self.rdi).count()
+        individuals_count = ImportedIndividual.objects.filter(registration_data_import=self.rdi).count()
+        self.assertEqual(households_count, 2)
+        self.assertEqual(individuals_count, 3)
+
+        data = {"delegates": [{"delegate_id": people_ids[2], "delegated_for": [people_ids[1]]}]}
+
+        response = self.client.post(self.url, data, format="json")
+        self.assertEqual(response.status_code, status.HTTP_200_OK, str(response.json()))
+        data = response.json()
+        self.assertEqual(data["updated"], 1)
+
+        hh1 = ImportedHousehold.objects.get(registration_data_import=self.rdi, village="village1")
+        hh2 = ImportedHousehold.objects.get(registration_data_import=self.rdi, village="village2")
+
+        self.assertEqual(hh1.primary_collector.full_name, "John Doe")
+        self.assertEqual(hh2.primary_collector.full_name, "Jack Jones")
+
+    def _create_people(self) -> List[UUID]:
+        data = [
+            {
+                "residence_status": "IDP",
+                "village": "village1",
+                "country": "AF",
+                "collect_individual_data": COLLECT_TYPE_FULL,
+                "full_name": "John Doe",
+                "birth_date": "2000-01-01",
+                "sex": "MALE",
+                "type": "",
+                "documents": [
+                    {
+                        "document_number": "10",
+                        "image": "",
+                        "doc_date": "2010-01-01",
+                        "country": "AF",
+                        "type": IDENTIFICATION_TYPE_TO_KEY_MAPPING[IDENTIFICATION_TYPE_BIRTH_CERTIFICATE],
+                    }
+                ],
+            },
+            {
+                "residence_status": "IDP",
+                "village": "village2",
+                "country": "AF",
+                "collect_individual_data": COLLECT_TYPE_FULL,
+                "full_name": "Mary Doe",
+                "birth_date": "1990-01-01",
+                "sex": "FEMALE",
+                "type": "",
+            },
+            {
+                "residence_status": "IDP",
+                "village": "village3",
+                "country": "AF",
+                "collect_individual_data": COLLECT_TYPE_FULL,
+                "full_name": "Jack Jones",
+                "birth_date": "1980-01-01",
+                "sex": "MALE",
+                "type": NON_BENEFICIARY,
+            },
+        ]
+        url = reverse("api:rdi-push-people", args=[self.business_area.slug, str(self.rdi.id)])
+        response = self.client.post(url, data, format="json")
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED, str(response.json()))
+        response_json = response.json()
+        return response_json["people"]