no message

Author: johniak

src/hct_mis_api/apps/core/utils.py

@@ -46,13 +46,15 @@
     from openpyxl.worksheet.worksheet import Worksheet
 
     from hct_mis_api.apps.account.models import User
+    from hct_mis_api.apps.geo.models import Area
+    from hct_mis_api.apps.household.models import Household
 
 logger = logging.getLogger(__name__)
 
 
 class CaseInsensitiveTuple(tuple):
     def __contains__(  # type: ignore # FIXME Signature of "__contains__" incompatible with supertype tuple
-            self, key: str, *args: Any, **kwargs: Any
+        self, key: str, *args: Any, **kwargs: Any
     ) -> bool:
         return key.casefold() in (element.casefold() for element in self)
 
@@ -92,11 +94,11 @@ def get_program_id_from_headers(headers: Union[Dict, "HttpHeaders"]) -> Optional
 
 
 def unique_slugify(
-        instance: "Model",
-        value: Any,
-        slug_field_name: str = "slug",
-        queryset: Optional["QuerySet"] = None,
-        slug_separator: str = "-",
+    instance: "Model",
+    value: Any,
+    slug_field_name: str = "slug",
+    queryset: Optional["QuerySet"] = None,
+    slug_separator: str = "-",
 ) -> None:
     """
     Calculates and stores a unique slug of ``value`` for an instance.
@@ -323,7 +325,7 @@ def encode_ids(results: Any, model_name: str, key: str) -> List[Dict]:
 
 
 def to_dict(
-        instance: "Model", fields: Union[List, Tuple, None] = None, dict_fields: Optional[Dict] = None
+    instance: "Model", fields: Union[List, Tuple, None] = None, dict_fields: Optional[Dict] = None
 ) -> Dict[str, Any]:
     from django.db.models import Model
     from django.forms import model_to_dict
@@ -590,11 +592,11 @@ def xlrd_rows_iterator(sheet: "Worksheet") -> Generator:
 
 
 def chart_get_filtered_qs(
-        qs: Any,
-        year: int,
-        business_area_slug_filter: Optional[Dict] = None,
-        additional_filters: Optional[Dict] = None,
-        year_filter_path: Optional[str] = None,
+    qs: Any,
+    year: int,
+    business_area_slug_filter: Optional[Dict] = None,
+    additional_filters: Optional[Dict] = None,
+    year_filter_path: Optional[str] = None,
 ) -> "QuerySet":
     if additional_filters is None:
         additional_filters = {}
@@ -626,7 +628,7 @@ def sum_lists_with_values(qs_values: Iterable, list_len: int) -> List[int]:
 
 
 def chart_permission_decorator(
-        chart_resolve: Optional[Callable] = None, permissions: Optional[List] = None
+    chart_resolve: Optional[Callable] = None, permissions: Optional[List] = None
 ) -> Callable:
     if chart_resolve is None:
         return functools.partial(chart_permission_decorator, permissions=permissions)
@@ -641,7 +643,7 @@ def resolve_f(*args: Any, **kwargs: Any) -> Any:
             business_area = BusinessArea.objects.filter(slug=business_area_slug).first()
             program_id = get_program_id_from_headers(resolve_info.context.headers)
             if any(
-                    resolve_info.context.user.has_permission(per.name, business_area, program_id) for per in permissions
+                resolve_info.context.user.has_permission(per.name, business_area, program_id) for per in permissions
             ):
                 return chart_resolve(*args, **kwargs)
             log_and_raise("Permission Denied")
@@ -657,7 +659,7 @@ def chart_filters_decoder(filters: Dict) -> Dict:
 
 
 def chart_create_filter_query(
-        filters: Dict, program_id_path: str = "id", administrative_area_path: str = "admin_areas"
+    filters: Dict, program_id_path: str = "id", administrative_area_path: str = "admin_areas"
 ) -> Dict:
     filter_query = {}
     if program := filters.get("program"):
@@ -673,7 +675,7 @@ def chart_create_filter_query(
 
 
 def chart_create_filter_query_for_payment_verification_gfk(
-        filters: Dict, program_id_path: str = "id", administrative_area_path: str = "admin_areas"
+    filters: Dict, program_id_path: str = "id", administrative_area_path: str = "admin_areas"
 ) -> Q:
     filter_query = Q()
     if program := filters.get("program"):
@@ -809,7 +811,7 @@ def timezone_datetime(value: Any) -> datetime:
 
 
 def save_data_in_cache(
-        cache_key: str, data_lambda: Callable, timeout: int = 60 * 60 * 24, cache_condition: Optional[Callable] = None
+    cache_key: str, data_lambda: Callable, timeout: int = 60 * 60 * 24, cache_condition: Optional[Callable] = None
 ) -> Any:
     cache_data = cache.get(cache_key, "NOT_CACHED")
     if cache_data == "NOT_CACHED":
@@ -882,7 +884,7 @@ def clear_cache_for_key(key: str) -> None:
 def chunks(lst: list, n: int) -> list:
     """Yield successive n-sized chunks from lst."""
     for i in range(0, len(lst), n):
-        yield lst[i: i + n]
+        yield lst[i : i + n]
 
 
 def send_email_notification_on_commit(service: Any, user: "User") -> None:
@@ -897,9 +899,9 @@ def send_email_notification_on_commit(service: Any, user: "User") -> None:
 
 
 def send_email_notification(
-        service: Any,
-        user: Optional["User"] = None,
-        context_kwargs: Optional[Dict] = None,
+    service: Any,
+    user: Optional["User"] = None,
+    context_kwargs: Optional[Dict] = None,
 ) -> None:
     if context_kwargs:
         context = service.get_email_context(**context_kwargs)

src/hct_mis_api/apps/household/models.py

@@ -594,7 +594,7 @@ def unwithdraw(self) -> None:
         self.withdrawn_date = None
         self.save()
 
-    def set_admin_areas(self, new_admin_area: Area, save: bool = True) -> None:
+    def set_admin_areas(self, new_admin_area: Optional[Area], save: bool = True) -> None:
         """Propagates admin1,2,3,4 based on admin_area parents"""
         admins = ["admin1", "admin2", "admin3", "admin4"]
         if not new_admin_area:
@@ -609,7 +609,7 @@ def set_admin_areas(self, new_admin_area: Area, save: bool = True) -> None:
             new_admin_area = getattr(new_admin_area, "parent", None)
 
         if save:
-            self.save(update_fields= admins)
+            self.save(update_fields=admins)
 
     @property
     def sanction_list_possible_match(self) -> bool:

src/hct_mis_api/apps/household/schema.py

@@ -42,12 +42,14 @@
     chart_filters_decoder,
     chart_permission_decorator,
     encode_ids,
+    get_lowest_admin_area,
     get_model_choices_fields,
     get_program_id_from_headers,
     resolve_flex_fields_choices_to_string,
     sum_lists_with_values,
-    to_choice_object, get_lowest_admin_area,
+    to_choice_object,
 )
+from hct_mis_api.apps.geo.models import Area
 from hct_mis_api.apps.geo.schema import AreaNode
 from hct_mis_api.apps.grievance.models import GrievanceTicket
 from hct_mis_api.apps.household.filters import (
@@ -345,9 +347,9 @@ def resolve_phone_no_alternative_valid(parent, info: Any) -> Boolean:
     def resolve_delivery_mechanisms_data(parent, info: Any) -> QuerySet[DeliveryMechanismData]:
         program_id = get_program_id_from_headers(info.context.headers)
         if not info.context.user.has_permission(
-                Permissions.POPULATION_VIEW_INDIVIDUAL_DELIVERY_MECHANISMS_SECTION.value,
-                parent.business_area,
-                program_id,
+            Permissions.POPULATION_VIEW_INDIVIDUAL_DELIVERY_MECHANISMS_SECTION.value,
+            parent.business_area,
+            program_id,
         ):
             return parent.delivery_mechanisms_data.none()
 
@@ -378,9 +380,9 @@ def check_node_permission(cls, info: Any, object_instance: Individual) -> None:
 
         # if user can't simply view all individuals, we check if they can do it because of grievance or rdi details
         if not user.has_permission(
-                Permissions.POPULATION_VIEW_INDIVIDUALS_DETAILS.value,
-                object_instance.business_area,
-                object_instance.program_id,
+            Permissions.POPULATION_VIEW_INDIVIDUALS_DETAILS.value,
+            object_instance.business_area,
+            object_instance.program_id,
         ) and not user.has_permission(
             Permissions.RDI_VIEW_DETAILS.value,
             object_instance.business_area,
@@ -469,8 +471,8 @@ def resolve_sanction_list_confirmed_match(parent: Household, info: Any) -> bool:
         return parent.sanction_list_confirmed_match
 
     @staticmethod
-    def resolve_admin_area(self, info: Any) -> AreaNode:
-        return get_lowest_admin_area(self)
+    def resolve_admin_area(parent: Household, info: Any) -> Optional[Area]:
+        return get_lowest_admin_area(parent)
 
     @staticmethod
     def resolve_admin_area_title(parent: Household, info: Any) -> str:
@@ -555,9 +557,9 @@ def check_node_permission(cls, info: Any, object_instance: Household) -> None:
 
         # if user doesn't have permission to view all households or RDI details, we check based on their grievance tickets
         if not user.has_permission(
-                Permissions.POPULATION_VIEW_HOUSEHOLDS_DETAILS.value,
-                object_instance.business_area,
-                object_instance.program_id,
+            Permissions.POPULATION_VIEW_HOUSEHOLDS_DETAILS.value,
+            object_instance.business_area,
+            object_instance.program_id,
         ) and not user.has_permission(
             Permissions.RDI_VIEW_DETAILS.value,
             object_instance.business_area,
@@ -816,16 +818,19 @@ def resolve_all_households(self, info: Any, **kwargs: Any) -> QuerySet:
 
             queryset = queryset.filter(filter_q)
 
-        if does_path_exist_in_query("edges.node.admin1", info) or does_path_exist_in_query("edges.node.admin_area",
-                                                                                           info):
+        if does_path_exist_in_query("edges.node.admin1", info) or does_path_exist_in_query(
+            "edges.node.admin_area", info
+        ):
             queryset = queryset.select_related("admin1")
             queryset = queryset.select_related("admin1__area_type")
-        if does_path_exist_in_query("edges.node.admin2", info) or does_path_exist_in_query("edges.node.admin_area",
-                                                                                           info):
+        if does_path_exist_in_query("edges.node.admin2", info) or does_path_exist_in_query(
+            "edges.node.admin_area", info
+        ):
             queryset = queryset.select_related("admin2")
             queryset = queryset.select_related("admin2__area_type")
-        if does_path_exist_in_query("edges.node.admin3", info) or does_path_exist_in_query("edges.node.admin_area",
-                                                                                           info):
+        if does_path_exist_in_query("edges.node.admin3", info) or does_path_exist_in_query(
+            "edges.node.admin_area", info
+        ):
             queryset = queryset.select_related("admin3")
             queryset = queryset.select_related("admin3__area_type")
 
@@ -899,7 +904,7 @@ def resolve_deduplication_status_choices(self, info: Any, **kwargs: Any) -> List
     @chart_permission_decorator(permissions=[Permissions.DASHBOARD_VIEW_COUNTRY])
     @cached_in_django_cache(24)
     def resolve_section_households_reached(
-            self, info: Any, business_area_slug: str, year: int, **kwargs: Any
+        self, info: Any, business_area_slug: str, year: int, **kwargs: Any
     ) -> Dict[str, int]:
         payment_items_qs: "QuerySet" = get_payment_items_for_dashboard(
             year, business_area_slug, chart_filters_decoder(kwargs), True
@@ -909,7 +914,7 @@ def resolve_section_households_reached(
     @chart_permission_decorator(permissions=[Permissions.DASHBOARD_VIEW_COUNTRY])
     @cached_in_django_cache(24)
     def resolve_section_individuals_reached(
-            self, info: Any, business_area_slug: str, year: int, **kwargs: Any
+        self, info: Any, business_area_slug: str, year: int, **kwargs: Any
     ) -> Dict[str, int]:
         payment_items_qs: "QuerySet" = get_payment_items_for_dashboard(
             year, business_area_slug, chart_filters_decoder(kwargs), True
@@ -920,7 +925,7 @@ def resolve_section_individuals_reached(
     @chart_permission_decorator(permissions=[Permissions.DASHBOARD_VIEW_COUNTRY])
     @cached_in_django_cache(24)
     def resolve_section_people_reached(
-            self, info: Any, business_area_slug: str, year: int, **kwargs: Any
+        self, info: Any, business_area_slug: str, year: int, **kwargs: Any
     ) -> Dict[str, int]:
         payment_items_qs: "QuerySet" = get_payment_items_for_dashboard(
             year, business_area_slug, chart_filters_decoder(kwargs), True
@@ -931,7 +936,7 @@ def resolve_section_people_reached(
     @chart_permission_decorator(permissions=[Permissions.DASHBOARD_VIEW_COUNTRY])
     @cached_in_django_cache(24)
     def resolve_section_child_reached(
-            self, info: Any, business_area_slug: str, year: int, **kwargs: Any
+        self, info: Any, business_area_slug: str, year: int, **kwargs: Any
     ) -> Dict[str, int]:
         households_child_params = [
             "female_age_group_0_5_count",
@@ -951,7 +956,7 @@ def resolve_section_child_reached(
     @chart_permission_decorator(permissions=[Permissions.DASHBOARD_VIEW_COUNTRY])
     @cached_in_django_cache(24)
     def resolve_chart_individuals_reached_by_age_and_gender(
-            self, info: Any, business_area_slug: str, year: int, **kwargs: Any
+        self, info: Any, business_area_slug: str, year: int, **kwargs: Any
     ) -> Dict:
         households_params = [
             "female_age_group_0_5_count",
@@ -979,7 +984,7 @@ def resolve_chart_individuals_reached_by_age_and_gender(
     @chart_permission_decorator(permissions=[Permissions.DASHBOARD_VIEW_COUNTRY])
     @cached_in_django_cache(24)
     def resolve_chart_people_reached_by_age_and_gender(
-            self, info: Any, business_area_slug: str, year: int, **kwargs: Any
+        self, info: Any, business_area_slug: str, year: int, **kwargs: Any
     ) -> Dict:
         households_params = [
             "female_age_group_0_5_count",
@@ -1007,7 +1012,7 @@ def resolve_chart_people_reached_by_age_and_gender(
     @chart_permission_decorator(permissions=[Permissions.DASHBOARD_VIEW_COUNTRY])
     @cached_in_django_cache(24)
     def resolve_chart_individuals_with_disability_reached_by_age(
-            self, info: Any, business_area_slug: str, year: int, **kwargs: Any
+        self, info: Any, business_area_slug: str, year: int, **kwargs: Any
     ) -> Dict:
         households_params_with_disability = [
             "female_age_group_0_5_disabled_count",
@@ -1069,7 +1074,7 @@ def resolve_chart_individuals_with_disability_reached_by_age(
     @chart_permission_decorator(permissions=[Permissions.DASHBOARD_VIEW_COUNTRY])
     @cached_in_django_cache(24)
     def resolve_chart_people_with_disability_reached_by_age(
-            self, info: Any, business_area_slug: str, year: int, **kwargs: Any
+        self, info: Any, business_area_slug: str, year: int, **kwargs: Any
     ) -> Dict:
         households_params_with_disability = [
             "female_age_group_0_5_disabled_count",

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

@@ -420,7 +420,6 @@ def handle_household(
         household_obj.registration_data_import = self.registration_data_import
         household_obj.program = self.registration_data_import.program
         household_obj.business_area = self.business_area
-        household_obj.set_admin_areas()
         households_to_create.append(household_obj)
         for ind in current_individuals:
             ind.first_registration_date = registration_date

tests/selenium/grievance/feedback/test_feedback.py

@@ -92,7 +92,6 @@ def create_households_and_individuals() -> Household:
     hh.admin1 = Area.objects.first()
     hh.admin2 = Area.objects.get(name="Shakardara")
     hh.save()
-    hh.set_admin_areas()
     hh.refresh_from_db()
     yield hh
 

tests/unit/apps/household/test_models.py

@@ -56,11 +56,10 @@ def setUpTestData(cls) -> None:
 
     def test_household_admin_areas_set(self) -> None:
         household, (individual) = create_household(household_args={"size": 1, "business_area": self.business_area})
-        household.admin_area = self.area1
         household.admin1 = self.area1
         household.save()
 
-        household.set_admin_areas()
+        household.set_admin_areas(self.area1)
         household.refresh_from_db()
 
         self.assertEqual(household.admin_area, self.area1)

tests/unit/apps/payment/test_finish_verification_plan.py

@@ -62,15 +62,14 @@ def setUpTestData(cls) -> None:
             household, _ = create_household(
                 {
                     "registration_data_import": registration_data_import,
-                    "admin_area": afghanistan_areas_qs.order_by("?").first(),
+                    "admin1": afghanistan_areas_qs.order_by("?").first(),
                     "program": cls.program,
                 },
                 {
                     "registration_data_import": registration_data_import,
                     "phone_no": f"+48 609 999 {i:03d}",
                 },
             )
-            household.set_admin_areas()
             household.program = cls.program
             household.refresh_from_db()