Unique constraint for unicef_id in program

pyproject.toml

@@ -137,7 +137,7 @@ distribution = true
 
 [project]
 name = "hope"
-version = "2.15.0"
+version = "2.16.0"
 description = "HCT MIS is UNICEF's humanitarian cash transfer platform."
 authors = [
     { name = "Tivix" },

src/frontend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "frontend",
-  "version": "2.15.0",
+  "version": "2.16.0",
   "private": true,
   "type": "module",
   "scripts": {

src/hct_mis_api/api/endpoints/rdi/upload.py

@@ -140,6 +140,7 @@ class Meta:
             "updated_at",
             "version",
             "vector_column",
+            "unicef_id",
         ]
 
     def validate_role(self, value: str) -> Optional[str]:
@@ -175,6 +176,7 @@ class Meta:
             "geopoint",
             "detail_id",
             "version",
+            "unicef_id",
         ]
         validators = [HouseholdValidator()]
 

src/hct_mis_api/apps/household/migrations/0005_migration.py

@@ -0,0 +1,21 @@
+# Generated by Django 3.2.25 on 2024-12-19 11:34
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('household', '0004_migration'),
+    ]
+
+    operations = [
+        migrations.AddConstraint(
+            model_name='household',
+            constraint=models.UniqueConstraint(condition=models.Q(('is_removed', False)), fields=('unicef_id', 'program'), name='unique_hh_unicef_id_in_program'),
+        ),
+        migrations.AddConstraint(
+            model_name='individual',
+            constraint=models.UniqueConstraint(condition=models.Q(('is_removed', False), ('duplicate', False)), fields=('unicef_id', 'program'), name='unique_ind_unicef_id_in_program'),
+        ),
+    ]

src/hct_mis_api/apps/household/models.py

@@ -576,6 +576,13 @@ class CollectType(models.TextChoices):
     class Meta:
         verbose_name = "Household"
         permissions = (("can_withdrawn", "Can withdrawn Household"),)
+        constraints = [
+            UniqueConstraint(
+                fields=["unicef_id", "program"],
+                condition=Q(is_removed=False),
+                name="unique_hh_unicef_id_in_program",
+            )
+        ]
 
     def save(self, *args: Any, **kwargs: Any) -> None:
         from hct_mis_api.apps.targeting.models import (
@@ -1187,6 +1194,13 @@ def __str__(self) -> str:
     class Meta:
         verbose_name = "Individual"
         indexes = (GinIndex(fields=["vector_column"]),)
+        constraints = [
+            UniqueConstraint(
+                fields=["unicef_id", "program"],
+                condition=Q(is_removed=False) & Q(duplicate=False),
+                name="unique_ind_unicef_id_in_program",
+            )
+        ]
 
     def recalculate_data(self, save: bool = True) -> Tuple[Any, List[str]]:
         update_fields = ["disability"]

src/hct_mis_api/apps/registration_datahub/mutations.py

@@ -115,15 +115,21 @@ def create_registration_data_import_for_import_program_population(
     pull_pictures = registration_data_import_data.pop("pull_pictures", True)
     screen_beneficiary = registration_data_import_data.pop("screen_beneficiary", False)
     import_from_program_id = registration_data_import_data.pop("import_from_program_id", None)
+    households_to_exclude = Household.all_merge_status_objects.filter(
+        program=import_to_program_id,
+    ).values_list("unicef_id", flat=True)
     households = Household.objects.filter(
         program_id=import_from_program_id,
         withdrawn=False,
-    ).exclude(household_collection__households__program=import_to_program_id)
+    ).exclude(unicef_id__in=households_to_exclude)
+    individuals_to_exclude = Individual.all_merge_status_objects.filter(
+        program=import_to_program_id,
+    ).values_list("unicef_id", flat=True)
     individuals = Individual.objects.filter(
         program_id=import_from_program_id,
         withdrawn=False,
         duplicate=False,
-    ).exclude(individual_collection__individuals__program=import_to_program_id)
+    ).exclude(unicef_id__in=individuals_to_exclude)
     created_obj_hct = RegistrationDataImport(
         status=RegistrationDataImport.IMPORTING,
         imported_by=user,

src/hct_mis_api/apps/registration_datahub/tasks/import_program_population.py

@@ -8,17 +8,23 @@
 def import_program_population(
     import_from_program_id: str, import_to_program_id: str, rdi: RegistrationDataImport
 ) -> None:
+    households_to_exclude = Household.all_merge_status_objects.filter(
+        program=import_to_program_id,
+    ).values_list("unicef_id", flat=True)
     copy_from_households = Household.objects.filter(
         program=import_from_program_id,
         withdrawn=False,
-    ).exclude(household_collection__households__program_id=import_to_program_id)
+    ).exclude(unicef_id__in=households_to_exclude)
+    individuals_to_exclude = Individual.all_merge_status_objects.filter(
+        program=import_to_program_id,
+    ).values_list("unicef_id", flat=True)
     copy_from_individuals = (
         Individual.objects.filter(
             program_id=import_from_program_id,
             withdrawn=False,
             duplicate=False,
         )
-        .exclude(individual_collection__individuals__program_id=import_to_program_id)
+        .exclude(unicef_id__in=individuals_to_exclude)
         .order_by("first_registration_date")
     )
     import_to_program = Program.objects.get(id=import_to_program_id)

tests/selenium/targeting/test_targeting.py

@@ -164,11 +164,13 @@ def create_flexible_attribute(
     return flexible_attribute
 
 
-def create_custom_household(observed_disability: list[str], residence_status: str = HOST) -> Household:
+def create_custom_household(
+    observed_disability: list[str], residence_status: str = HOST, unicef_id: str = "HH-00-0000.0442"
+) -> Household:
     program = Program.objects.get(name="Test Programm")
     household, _ = create_household_and_individuals(
         household_data={
-            "unicef_id": "HH-00-0000.0442",
+            "unicef_id": unicef_id,
             "rdi_merge_status": "MERGED",
             "business_area": program.business_area,
             "program": program,
@@ -187,17 +189,17 @@ def create_custom_household(observed_disability: list[str], residence_status: st
 
 @pytest.fixture
 def household_with_disability() -> Household:
-    yield create_custom_household(observed_disability=[SEEING, HEARING])
+    yield create_custom_household(observed_disability=[SEEING, HEARING], unicef_id="HH-00-0000.0443")
 
 
 @pytest.fixture
 def household_without_disabilities() -> Household:
-    yield create_custom_household(observed_disability=[])
+    yield create_custom_household(observed_disability=[], unicef_id="HH-00-0000.0444")
 
 
 @pytest.fixture
 def household_refugee() -> Household:
-    yield create_custom_household(observed_disability=[], residence_status=REFUGEE)
+    yield create_custom_household(observed_disability=[], residence_status=REFUGEE, unicef_id="HH-00-0000.0445")
 
 
 def get_program_with_dct_type_and_name(
@@ -235,7 +237,7 @@ def create_targeting(household_without_disabilities: Household) -> TargetPopulat
     target_population.save()
     household, _ = create_household(
         household_args={
-            "unicef_id": "HH-00-0000.0442",
+            "unicef_id": "HH-00-0000.0440",
             "business_area": program.business_area,
             "program": program,
             "residence_status": HOST,

tests/unit/apps/household/test_models.py

@@ -7,7 +7,12 @@
 from hct_mis_api.apps.core.utils import IDENTIFICATION_TYPE_TO_KEY_MAPPING
 from hct_mis_api.apps.geo.fixtures import AreaFactory, AreaTypeFactory
 from hct_mis_api.apps.geo.models import Country
-from hct_mis_api.apps.household.fixtures import BankAccountInfoFactory, create_household
+from hct_mis_api.apps.household.fixtures import (
+    BankAccountInfoFactory,
+    HouseholdFactory,
+    IndividualFactory,
+    create_household,
+)
 from hct_mis_api.apps.household.models import (
     IDENTIFICATION_TYPE_NATIONAL_PASSPORT,
     IDENTIFICATION_TYPE_OTHER,
@@ -30,6 +35,7 @@ def setUpTestData(cls) -> None:
         super().setUpTestData()
         create_afghanistan()
         cls.business_area = BusinessArea.objects.get(slug="afghanistan")
+        cls.program = ProgramFactory(business_area=cls.business_area)
 
         area_type_level_1 = AreaTypeFactory(
             name="State1",
@@ -114,15 +120,21 @@ def test_remove_household(self) -> None:
         household2.delete(soft=False)
         self.assertIsNone(Household.all_objects.filter(unicef_id="HH-9191").first())
 
+    def test_unique_unicef_id_per_program_constraint(self) -> None:
+        HouseholdFactory(unicef_id="HH-123", program=self.program)
+        HouseholdFactory(unicef_id="HH-000", program=self.program)
+        with self.assertRaises(IntegrityError):
+            HouseholdFactory(unicef_id="HH-123", program=self.program)
+
 
 class TestDocument(TestCase):
     @classmethod
     def setUpTestData(cls) -> None:
         super().setUpTestData()
         call_command("loadcountries")
-        business_area = create_afghanistan()
+        cls.business_area = create_afghanistan()
         afghanistan = Country.objects.get(name="Afghanistan")
-        _, (individual,) = create_household(household_args={"size": 1, "business_area": business_area})
+        _, (individual,) = create_household(household_args={"size": 1, "business_area": cls.business_area})
 
         cls.country = afghanistan
         cls.individual = individual
@@ -187,10 +199,9 @@ def test_create_representation_with_the_same_number(self) -> None:
         program_3 = ProgramFactory()
         program_4 = ProgramFactory()
 
-        for _program in [program_1, program_2]:
-            (individual_to_create, documents_to_create, _, _) = copy_individual_fast(self.individual, _program)
-            Individual.objects.bulk_create([individual_to_create])
-            Document.objects.bulk_create(documents_to_create)
+        (individual_to_create, documents_to_create, _, _) = copy_individual_fast(self.individual, program_2)
+        Individual.objects.bulk_create([individual_to_create])
+        Document.objects.bulk_create(documents_to_create)
 
         # test regular create
         for _program in [program_3, program_4]:
@@ -209,16 +220,17 @@ def test_create_representation_with_the_same_number(self) -> None:
             )
 
         # don't allow to create representations with the same document number and programs
-        (individual_to_create, _, _, _) = copy_individual_fast(self.individual, _program)
-        (created_individual_representation,) = Individual.objects.bulk_create([individual_to_create])
+        _, (individual,) = create_household(
+            household_args={"size": 1, "business_area": self.business_area, "program": program_1}
+        )
         with self.assertRaises(IntegrityError):
             with transaction.atomic():
                 # bulk create
                 Document.objects.bulk_create(
                     [
                         Document(
                             document_number="213123",
-                            individual=created_individual_representation,
+                            individual=individual,
                             country=self.country,
                             type=document_type,
                             status=Document.STATUS_VALID,
@@ -234,7 +246,7 @@ def test_create_representation_with_the_same_number(self) -> None:
                 # regular create
                 Document.objects.create(
                     document_number="213123",
-                    individual=created_individual_representation,
+                    individual=individual,
                     country=self.country,
                     type=document_type,
                     status=Document.STATUS_VALID,
@@ -421,21 +433,16 @@ def test_create_representations_duplicated_documents_with_different_numbers_and_
         # allow to create representations with the same document number within different programs
         self.individual.is_original = True
         self.individual.save()
-
-        program_1 = self.individual.program
         program_2 = ProgramFactory()
         program_3 = ProgramFactory()
 
-        # make representations with the same number
-        for _program in [program_1, program_2]:
-            (individual_to_create, documents_to_create, _, _) = copy_individual_fast(self.individual, _program)
-            Individual.objects.bulk_create([individual_to_create])
-            Document.objects.bulk_create(documents_to_create)
+        # make representation with the same number
+        (individual_to_create, documents_to_create, _, _) = copy_individual_fast(self.individual, program_2)
+        Individual.objects.bulk_create([individual_to_create])
+        Document.objects.bulk_create(documents_to_create)
 
         # make representation with different number
-        program_3_individual_representation = (individual_to_create, _, _, _) = copy_individual_fast(
-            self.individual, program_3
-        )
+        (individual_to_create, _, _, _) = copy_individual_fast(self.individual, program_3)
         (program_3_individual_representation,) = Individual.objects.bulk_create([individual_to_create])
         Document.objects.create(
             document_number="456",
@@ -509,8 +516,8 @@ class TestIndividualModel(TestCase):
     @classmethod
     def setUpTestData(cls) -> None:
         super().setUpTestData()
-        create_afghanistan()
-        ProgramFactory()
+        business_area = create_afghanistan()
+        cls.program = ProgramFactory(business_area=business_area)
 
     def test_bank_name(self) -> None:
         individual = create_household({"size": 1})[1][0]
@@ -531,3 +538,9 @@ def test_bank_branch_name(self) -> None:
         individual = create_household({"size": 1})[1][0]
         bank_account_info = BankAccountInfoFactory(individual=individual)
         self.assertEqual(individual.bank_branch_name, bank_account_info.bank_branch_name)
+
+    def test_unique_unicef_id_per_program_constraint(self) -> None:
+        IndividualFactory(unicef_id="IND-123", program=self.program)
+        IndividualFactory(unicef_id="IND-000", program=self.program)
+        with self.assertRaises(IntegrityError):
+            IndividualFactory(unicef_id="IND-123", program=self.program)