add ! External Beneficiary Reference field

vitali-yanushchyk-valor · vitali-yanushchyk-valor · commit 57bf55fd8cd4 · 2025-07-31T09:28:15.000-04:00
diff --git a/src/country_workspace/contrib/hope/apps.py b/src/country_workspace/contrib/hope/apps.py
@@ -3,6 +3,7 @@
 from .geo import Admin1Choice, Admin2Choice, Admin3Choice, Admin4Choice, CountryChoice
 from .lookups import FinancialInstitutionChoice
 from .phone_numbers import PhoneNumberField
+from .beneficiary_reference import BeneficiaryReferenceModelChoice
 
 
 class Config(AppConfig):
@@ -23,6 +24,7 @@ def ready(self) -> None:
         field_registry.register(Admin4Choice)
         field_registry.register(FinancialInstitutionChoice)
         field_registry.register(PhoneNumberField)
+        field_registry.register(BeneficiaryReferenceModelChoice)
 
         from country_workspace.contrib.hope.validators import FullHouseholdValidator
         from country_workspace.validators.registry import beneficiary_validator_registry
diff --git a/src/country_workspace/contrib/hope/beneficiary_reference.py b/src/country_workspace/contrib/hope/beneficiary_reference.py
@@ -0,0 +1,101 @@
+from typing import Any, TYPE_CHECKING
+from django.forms import ModelChoiceField
+from django_select2.forms import ModelSelect2Widget
+from django.db.models import QuerySet
+from django.core.exceptions import ValidationError
+from django.urls import resolve
+from country_workspace.state import state
+
+if TYPE_CHECKING:
+    from country_workspace.models import Individual
+
+
+def get_household_id_from_request() -> int | None:
+    if not state.request:
+        return None
+
+    if (resolved := resolve(state.request.path)) and "object_id" in resolved.kwargs:
+        return int(resolved.kwargs["object_id"])
+    return None
+
+
+class BeneficiarySelect2Widget(ModelSelect2Widget):
+    search_fields = ["name__icontains"]
+
+    def __init__(
+        self, program_id: int | None = None, household_id: int | None = None, *args: Any, **kwargs: Any
+    ) -> None:
+        self.program_id = program_id
+        self.household_id = household_id
+        kwargs.setdefault(
+            "attrs",
+            {
+                "data-minimum-input-length": 0,
+                "class": "form-control",
+            },
+        )
+        super().__init__(*args, **kwargs)
+
+    def get_queryset(self) -> QuerySet["Individual"]:
+        from django.apps import apps
+
+        Individual = apps.get_model("country_workspace", "Individual")
+
+        if not self.program_id:
+            return Individual.objects.none()
+
+        try:
+            Program = apps.get_model("country_workspace", "Program")
+            program = Program.objects.get(id=self.program_id)
+            queryset = program.individuals.filter(removed=False)
+            if self.household_id:
+                queryset = queryset.filter(household=self.household_id)
+        except Program.DoesNotExist:
+            return Individual.objects.none()
+        else:
+            return queryset
+
+
+class BeneficiaryReferenceModelChoice(ModelChoiceField):
+    def __init__(self, filter_by_household: bool = False, *args: Any, **kwargs: Any) -> None:
+        program = state.program
+        household_id = get_household_id_from_request() if filter_by_household else None
+
+        queryset = program.individuals.filter(removed=False)
+        if household_id:
+            queryset = queryset.filter(household=household_id)
+
+        kwargs.setdefault("queryset", queryset)
+        kwargs.setdefault(
+            "widget", BeneficiarySelect2Widget(program_id=program.id, household_id=household_id, model=queryset.model)
+        )
+
+        super().__init__(*args, **kwargs)
+
+    def to_python(self, value: Any) -> str | None:
+        if value in self.empty_values:
+            return None
+
+        match value:
+            case int() | str() if str(value).isdigit():
+                try:
+                    individual = self.queryset.get(pk=int(value))
+                except self.queryset.model.DoesNotExist:
+                    raise ValidationError(self.error_messages["invalid_choice"], code="invalid_choice")
+                return individual.name
+
+            case str() if self.queryset.filter(name=value).exists():
+                return value
+
+            case _:
+                raise ValidationError(self.error_messages["invalid_choice"], code="invalid_choice")
+
+    def prepare_value(self, value: Any) -> int | None:
+        if isinstance(value, str) and value:
+            try:
+                individual = self.queryset.get(name=value)
+            except self.queryset.model.DoesNotExist:
+                return None
+            else:
+                return individual.pk
+        return getattr(value, "pk", value)
diff --git a/src/country_workspace/versioning/scripts/0016_add_beneficiary_reference.py b/src/country_workspace/versioning/scripts/0016_add_beneficiary_reference.py
@@ -0,0 +1,72 @@
+# Generated by HCW 0.1.0 on 2025 07 29 07:39:07
+from typing import Any
+from packaging.version import Version
+from concurrency.utils import fqn
+from contextlib import suppress
+from django.db import transaction
+from django.utils.text import slugify
+
+from hope_flex_fields.models import FieldDefinition, Fieldset
+from hope_flex_fields.utils import get_kwargs_from_field_class, get_common_attrs
+from country_workspace.contrib.hope.beneficiary_reference import BeneficiaryReferenceModelChoice
+from country_workspace.contrib.hope.constants import HOUSEHOLD_FIELDSET_NAME
+
+_script_for_version = Version("0.1.0")
+
+
+attrs_default = lambda cls: get_kwargs_from_field_class(cls, get_common_attrs())
+
+FIELDS: dict[str, dict[str, Any]] = {
+    "alternate_collector": {
+        "name": "Alternate Collector Reference",
+        "defaults": {
+            "field_type": fqn(BeneficiaryReferenceModelChoice),
+            "attrs": {
+                **attrs_default(BeneficiaryReferenceModelChoice),
+                "filter_by_household": False,
+                "help_text": "A unique reference for an alternate collector.",
+            },
+        },
+    },
+    "primary_collector": {
+        "name": "Primary Collector Reference",
+        "defaults": {
+            "field_type": fqn(BeneficiaryReferenceModelChoice),
+            "attrs": {
+                **attrs_default(BeneficiaryReferenceModelChoice),
+                "filter_by_household": True,
+                "help_text": "A unique reference for a primary collector.",
+            },
+        },
+    },
+}
+
+
+def forward() -> None:
+    with transaction.atomic():
+        fs, __ = Fieldset.objects.get_or_create(name=HOUSEHOLD_FIELDSET_NAME)
+        for field_name, field_config in FIELDS.items():
+            field_config["defaults"]["slug"] = slugify(field_config["name"])
+            fd, __ = FieldDefinition.objects.update_or_create(
+                name=field_config["name"], defaults=field_config["defaults"]
+            )
+            fs.fields.update_or_create(name=field_name, defaults={"definition": fd})
+
+
+def backward() -> None:
+    with transaction.atomic():
+        try:
+            fs = Fieldset.objects.get(name=HOUSEHOLD_FIELDSET_NAME)
+        except Fieldset.DoesNotExist:
+            return
+
+        for field_name, field_config in FIELDS.items():
+            fs.fields.filter(name=field_name).delete()
+            with suppress(FieldDefinition.DoesNotExist):
+                fd = FieldDefinition.objects.get(name=field_config["name"])
+                fd.delete()
+
+
+class Scripts:
+    requires = []
+    operations = [(forward, backward)]
diff --git a/tests/contrib/hope/test_beneficiary_reference.py b/tests/contrib/hope/test_beneficiary_reference.py
@@ -0,0 +1,205 @@
+import pytest
+from unittest.mock import patch, MagicMock
+from django.core.exceptions import ValidationError
+
+from country_workspace.state import state
+from country_workspace.contrib.hope.beneficiary_reference import (
+    get_household_id_from_request,
+    BeneficiarySelect2Widget,
+    BeneficiaryReferenceModelChoice,
+)
+
+
+@pytest.fixture
+def office():
+    from testutils.factories import OfficeFactory
+
+    co = OfficeFactory()
+    state.tenant = co
+    return co
+
+
+@pytest.fixture
+def program(office):
+    from testutils.factories import CountryProgramFactory
+
+    return CountryProgramFactory(country_office=office)
+
+
+@pytest.fixture
+def household(program):
+    from testutils.factories import CountryHouseholdFactory
+
+    return CountryHouseholdFactory(batch__program=program)
+
+
+@pytest.fixture
+def individual(household):
+    return household.members.first()
+
+
+@pytest.fixture
+def multiple_households(program):
+    from testutils.factories import CountryHouseholdFactory
+
+    return [CountryHouseholdFactory(batch__program=program) for _ in range(3)]
+
+
+@pytest.fixture
+def all_individuals(multiple_households):
+    individuals = []
+    for hh in multiple_households:
+        individuals.extend(hh.members.all())
+    return individuals
+
+
+@pytest.mark.parametrize(
+    ("request_data", "expected"),
+    [
+        (None, None),
+        (MagicMock(kwargs={"object_id": "456"}), 456),
+        (MagicMock(kwargs={}), None),
+    ],
+)
+def test_get_household_id_from_request(request_data, expected):
+    with (
+        patch("country_workspace.contrib.hope.beneficiary_reference.state") as mock_state,
+        patch("country_workspace.contrib.hope.beneficiary_reference.resolve") as mock_resolve,
+    ):
+        if request_data is None:
+            mock_state.request = None
+        else:
+            mock_resolve.return_value = request_data
+
+        assert get_household_id_from_request() == expected
+
+
+@pytest.mark.parametrize(
+    ("program_id", "household_id"),
+    [
+        (None, None),
+        (123, None),
+        (123, 456),
+    ],
+)
+def test_beneficiary_widget_init(program_id, household_id):
+    widget = BeneficiarySelect2Widget(program_id=program_id, household_id=household_id)
+    assert widget.program_id == program_id
+    assert widget.household_id == household_id
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("program_id", [99999, None], ids=["non_existing_program", "no_program"])
+def test_widget_get_queryset_returns_empty_for_invalid_or_missing_program(program_id):
+    widget = BeneficiarySelect2Widget(program_id=program_id)
+    queryset = widget.get_queryset()
+    assert queryset.count() == 0
+
+
+@pytest.mark.django_db
+def test_widget_get_queryset_with_household_filter(program, household, individual):
+    widget = BeneficiarySelect2Widget(program_id=program.id, household_id=household.id)
+    assert individual in widget.get_queryset()
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("filter_by_household", [True, False])
+def test_field_init(program, household, individual, multiple_households, all_individuals, filter_by_household):
+    with (
+        patch("country_workspace.contrib.hope.beneficiary_reference.state") as mock_state,
+        patch("country_workspace.contrib.hope.beneficiary_reference.get_household_id_from_request") as mock_get,
+    ):
+        mock_state.program = program
+        mock_get.return_value = household.id if filter_by_household else None
+
+        field = BeneficiaryReferenceModelChoice(filter_by_household=filter_by_household)
+
+        if filter_by_household:
+            assert individual in field.queryset
+            assert all(ind.household == household for ind in field.queryset)
+        else:
+            for ind in all_individuals:
+                assert ind in field.queryset
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        (None, None),
+        ("", None),
+        ([], None),
+    ],
+)
+def test_to_python_empty_values(program, value, expected):
+    with patch("country_workspace.contrib.hope.beneficiary_reference.state") as mock_state:
+        mock_state.program = program
+        field = BeneficiaryReferenceModelChoice()
+        assert field.to_python(value) == expected
+
+
+@pytest.mark.django_db
+def test_to_python_valid_id(program, all_individuals):
+    with patch("country_workspace.contrib.hope.beneficiary_reference.state") as mock_state:
+        mock_state.program = program
+        field = BeneficiaryReferenceModelChoice()
+        individual = all_individuals[0]
+        assert field.to_python(str(individual.pk)) == individual.name
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    ("value", "should_raise"),
+    [
+        ("99999", True),  # Invalid ID
+        (99999, True),  # Invalid ID as int
+        ("Non Existing Name", True),  # Non-existing name
+        (["invalid", "list"], True),  # Invalid type
+    ],
+)
+def test_to_python_validation_errors(program, value, should_raise):
+    with patch("country_workspace.contrib.hope.beneficiary_reference.state") as mock_state:
+        mock_state.program = program
+        field = BeneficiaryReferenceModelChoice()
+
+        if should_raise:
+            with pytest.raises(ValidationError):
+                field.to_python(value)
+
+
+@pytest.mark.django_db
+def test_to_python_existing_name(program, all_individuals):
+    with patch("country_workspace.contrib.hope.beneficiary_reference.state") as mock_state:
+        mock_state.program = program
+        field = BeneficiaryReferenceModelChoice()
+        individual = all_individuals[0]
+
+        # Test with existing name - should return the same name
+        result = field.to_python(individual.name)
+        assert result == individual.name
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("input_type", ["pk_int", "name", "object", "non_existing_name", "empty_string"])
+def test_prepare_value(program, all_individuals, input_type):
+    with patch("country_workspace.contrib.hope.beneficiary_reference.state") as mock_state:
+        mock_state.program = program
+        field = BeneficiaryReferenceModelChoice()
+        individual = all_individuals[0]
+
+        match input_type:
+            case "pk_int":
+                result = field.prepare_value(individual.pk)
+                assert result == individual.pk
+            case "name":
+                result = field.prepare_value(individual.name)
+                assert result == individual.pk
+            case "object":
+                result = field.prepare_value(individual)
+                assert result == individual.pk
+            case "non_existing_name":
+                result = field.prepare_value("Non Existing Name")
+                assert result is None
+            case "empty_string":
+                result = field.prepare_value("")
+                assert result == ""
