chg ! tests

Author: vitali-yanushchyk-valor

src/country_workspace/contrib/hope/client.py

@@ -6,7 +6,7 @@
 
 import requests
 from constance import config
-from requests.exceptions import RequestException
+from requests.exceptions import RequestException, HTTPError
 
 from country_workspace.exceptions import RemoteError
 
@@ -69,21 +69,34 @@ def post(self, path: str, data: "JsonType | None") -> "FlatJsonType":
         url = self.get_url(path)
         signature = hashlib.sha256(f"{url}{data}{time.perf_counter_ns()}".encode()).hexdigest()
         hope_request_start.send(self.__class__, url=url, data=data, signature=signature)
+        response = None
+
         try:
-            ret = requests.post(
+            response = requests.post(
                 url,
                 json=data,
                 headers={"Authorization": f"Token {self.token}"},
                 timeout=10,  # nosec
             )
-            ret.raise_for_status()
-        except RequestException as exc:
-            raise RemoteError(f"Error posting to {url}: {exc}") from exc
+            response.raise_for_status()
+        except HTTPError as http_err:
+            resp = http_err.response
+            error_details = resp.text[:1000] + "..." if len(resp.text) > 1000 else resp.text
+            raise RemoteError(
+                f"HTTP error posting to {url}: {http_err}. "
+                f"Status Code: {resp.status_code}. Response Body: {error_details}"
+            ) from http_err
+        except RequestException as req_err:
+            raise RemoteError(f"Request failed for {url}: {req_err}") from req_err
 
         try:
-            result = ret.json()
-        except JSONDecodeError as exc:
-            raise RemoteError(f"Wrong JSON response posting to {url}") from exc
+            result = response.json()
+        except JSONDecodeError as json_err:
+            response_text = response.text if response else "N/A"
+            raise RemoteError(
+                f"Wrong JSON response posting to {url}. Status: {response.status_code}. Response text: {response_text}"
+            ) from json_err
 
         hope_request_end.send(self.__class__, url=url, data=data, signature=signature)
+
         return result

src/country_workspace/contrib/hope/push.py

@@ -13,6 +13,7 @@
 from country_workspace.exceptions import RemoteError
 from country_workspace.models import AsyncJob
 from country_workspace.workspaces.models import CountryHousehold, CountryIndividual
+from country_workspace.utils.fields import map_fields
 
 
 @dataclass
@@ -225,20 +226,3 @@ def steps() -> Iterator[Callable[[], None]]:
         if processor.total["errors"]:
             return processor.total
     return processor.total
-
-
-def map_fields(fields: dict[str, str]) -> dict[str, str]:
-    """
-    Map keys in a dictionary to alternative names based on a predefined mapping.
-
-    Args:
-        fields (dict[str, str]): A dictionary containing field names as keys and their values.
-
-    Returns:
-        dict[str, str]: A new dictionary with keys mapped according to the predefined mapping.
-
-    """
-    to_map = {
-        "gender": "sex",
-    }
-    return {to_map.get(k, k): v for k, v in fields.items()}

src/country_workspace/utils/fields.py

@@ -10,8 +10,9 @@
 Record = Mapping[str, Any]
 
 
-TO_REMOVE = "_h_c", "_h_f", "_i_c", "_i_f"
-TO_UPPERCASE = "relationship", "gender", "disability", "residence_status"
+TO_REMOVE_VALUES = "_h_c", "_h_f", "_i_c", "_i_f"
+TO_UPPERCASE_FIELDS = "relationship", "gender", "disability", "residence_status"
+TO_MAP_FIELDS = {"gender": "sex"}
 
 
 def clean_field_name(v: str) -> str:
@@ -24,7 +25,7 @@ def clean_field_name(v: str) -> str:
         str: The cleaned field name.
 
     """
-    return reduce(lambda name, substr: name.replace(substr, ""), TO_REMOVE, v.lower())
+    return reduce(lambda name, substr: name.replace(substr, ""), TO_REMOVE_VALUES, v.lower())
 
 
 def clean_field_names(record: Record) -> Record:
@@ -52,4 +53,18 @@ def uppercase_field_value(k: str, v: Any) -> str:
         str: The uppercase value if applicable or the original value.
 
     """
-    return v.upper() if isinstance(v, str) and any(k.startswith(prefix) for prefix in TO_UPPERCASE) else v
+    return v.upper() if isinstance(v, str) and any(k.startswith(prefix) for prefix in TO_UPPERCASE_FIELDS) else v
+
+
+def map_fields(fields: dict[str, str]) -> dict[str, str]:
+    """
+    Map keys in a dictionary to alternative names based on a predefined mapping.
+
+    Args:
+        fields (dict[str, str]): A dictionary containing field names as keys and their values.
+
+    Returns:
+        dict[str, str]: A new dictionary with keys mapped according to the predefined mapping.
+
+    """
+    return {TO_MAP_FIELDS.get(k, k): v for k, v in fields.items()}

src/country_workspace/versioning/checkers.py

@@ -1,3 +1,4 @@
+from typing import Any
 from django import forms
 from hope_flex_fields.models import DataChecker, FieldDefinition, Fieldset
 
@@ -7,40 +8,45 @@
     PEOPLE_CHECKER_NAME,
 )
 
+type FieldSpec = tuple[str, FieldDefinition, dict[str, Any] | None]
 
-def create_hope_checkers() -> None:
-    _char = FieldDefinition.objects.get(field_type=forms.CharField)
-    _date = FieldDefinition.objects.get(field_type=forms.DateField)
-    _bool = FieldDefinition.objects.get(field_type=forms.BooleanField)
-    _int = FieldDefinition.objects.get(field_type=forms.IntegerField)
-
-    _h_country = FieldDefinition.objects.get(name="CountryChoice")
-    _h_residence = FieldDefinition.objects.get(slug="hope-hh-residencestatus")
-    _i_gender = FieldDefinition.objects.get(slug="hope-ind-gender")
-    _i_disability = FieldDefinition.objects.get(slug="hope-ind-disability")
-    _i_role = FieldDefinition.objects.get(slug="hope-ind-role")
-    _i_relationship = FieldDefinition.objects.get(slug="hope-ind-relationship")
-
-    _p_type = FieldDefinition.objects.get(slug="hope-people-type")
 
-    hh_fs, __ = Fieldset.objects.get_or_create(name=HOUSEHOLD_CHECKER_NAME)
-    hh_fs.fields.get_or_create(name="address", definition=_char)
-    hh_fs.fields.get_or_create(name="admin1", definition=_char)
-    hh_fs.fields.get_or_create(name="admin2", definition=_char)
-    hh_fs.fields.get_or_create(name="admin3", definition=_char)
-    hh_fs.fields.get_or_create(name="admin4", definition=_char)
-    hh_fs.fields.get_or_create(name="collect_individual_data", definition=_bool)
-    hh_fs.fields.get_or_create(name="consent", definition=_bool)
-    hh_fs.fields.get_or_create(name="country", attrs={"label": "Country", "required": True}, definition=_h_country)
-    hh_fs.fields.get_or_create(name="country_origin", definition=_h_country)
-    hh_fs.fields.get_or_create(name="household_id", attrs={"label": "Household ID"}, definition=_char)
-    hh_fs.fields.get_or_create(name="name_enumerator", attrs={"label": "Enumerator"}, definition=_char)
-    hh_fs.fields.get_or_create(name="org_enumerator", definition=_char)
-    hh_fs.fields.get_or_create(name="registration_method", definition=_char)
-    hh_fs.fields.get_or_create(name="residence_status", definition=_h_residence)
-    hh_fs.fields.get_or_create(name="size", definition=_int)
+def create_hope_checkers() -> None:
+    try:
+        defs: dict[str, FieldDefinition] = {
+            "char": FieldDefinition.objects.get(field_type=forms.CharField),
+            "date": FieldDefinition.objects.get(field_type=forms.DateField),
+            "bool": FieldDefinition.objects.get(field_type=forms.BooleanField),
+            "int": FieldDefinition.objects.get(field_type=forms.IntegerField),
+            "h_country": FieldDefinition.objects.get(name="CountryChoice"),
+            "h_residence": FieldDefinition.objects.get(slug="hope-hh-residencestatus"),
+            "i_gender": FieldDefinition.objects.get(slug="hope-ind-gender"),
+            "i_disability": FieldDefinition.objects.get(slug="hope-ind-disability"),
+            "i_role": FieldDefinition.objects.get(slug="hope-ind-role"),
+            "i_relationship": FieldDefinition.objects.get(slug="hope-ind-relationship"),
+            "p_type": FieldDefinition.objects.get(slug="hope-people-type"),
+        }
+    except FieldDefinition.DoesNotExist as e:
+        raise LookupError(f"Could not find base FieldDefinitions needed for Hope checkers: {e}") from e
 
-    for segment in [
+    household_fields_spec: list[FieldSpec] = [
+        ("address", defs["char"], None),
+        ("admin1", defs["char"], None),
+        ("admin2", defs["char"], None),
+        ("admin3", defs["char"], None),
+        ("admin4", defs["char"], None),
+        ("collect_individual_data", defs["bool"], None),
+        ("consent", defs["bool"], None),
+        ("country", defs["h_country"], {"label": "Country", "required": True}),
+        ("country_origin", defs["h_country"], None),
+        ("household_id", defs["char"], {"label": "Household ID"}),
+        ("name_enumerator", defs["char"], {"label": "Enumerator"}),
+        ("org_enumerator", defs["char"], None),
+        ("registration_method", defs["char"], None),
+        ("residence_status", defs["h_residence"], None),
+        ("size", defs["int"], None),
+    ]
+    demographic_segments: list[str] = [
         "female_age_group_0_5_count",
         "female_age_group_6_11_count",
         "female_age_group_12_17_count",
@@ -62,58 +68,63 @@ def create_hope_checkers() -> None:
         "male_age_group_12_17_disabled_count",
         "male_age_group_18_59_disabled_count",
         "male_age_group_60_disabled_count",
-    ]:
-        hh_fs.fields.get_or_create(name=segment, definition=_int, attrs={"required": False})
+    ]
+    household_fields_spec.extend([(segment, defs["int"], {"required": False}) for segment in demographic_segments])
+
+    individual_fields_spec: list[FieldSpec] = [
+        ("address", defs["char"], None),
+        ("alternate_collector_id", defs["char"], {"label": "Alternative Collector for"}),
+        ("birth_date", defs["date"], {"label": "Birth Date", "required": True}),
+        ("disability", defs["i_disability"], {"label": "Disability"}),
+        ("estimated_birth_date", defs["bool"], {"label": "Estimated Birth Date", "required": False}),
+        ("family_name", defs["char"], {"label": "Family Name"}),
+        ("full_name", defs["char"], {"label": "Full Name", "required": True}),
+        ("gender", defs["i_gender"], None),
+        ("given_name", defs["char"], {"label": "Given Name"}),
+        ("middle_name", defs["char"], {"label": "Middle Name"}),
+        ("national_id_issuer", defs["char"], None),
+        ("national_id_no", defs["char"], None),
+        ("national_id_photo", defs["char"], None),
+        ("phone_no", defs["char"], None),
+        ("primary_collector_id", defs["char"], {"label": "Primary Collector for"}),
+        ("relationship", defs["i_relationship"], {"label": "Relationship", "required": True}),
+        ("role", defs["i_role"], {"label": "Role"}),
+    ]
 
-    ind_fs, __ = Fieldset.objects.get_or_create(name=INDIVIDUAL_CHECKER_NAME)
-    ind_fs.fields.get_or_create(name="address", definition=_char)
-    ind_fs.fields.get_or_create(
-        name="alternate_collector_id",
-        attrs={"label": "Alternative Collector for"},
-        definition=_char,
-    )
-    ind_fs.fields.get_or_create(name="birth_date", attrs={"label": "Birth Date", "required": True}, definition=_date)
-    ind_fs.fields.get_or_create(name="disability", attrs={"label": "Disability"}, definition=_i_disability)
-    ind_fs.fields.get_or_create(
-        name="estimated_birth_date", attrs={"label": "Estimated Birth Date", "required": False}, definition=_bool
-    )
-    ind_fs.fields.get_or_create(name="family_name", attrs={"label": "Family Name"}, definition=_char)
-    ind_fs.fields.get_or_create(name="full_name", attrs={"label": "Full Name", "required": True}, definition=_char)
-    ind_fs.fields.get_or_create(name="gender", definition=_i_gender)
-    ind_fs.fields.get_or_create(name="given_name", attrs={"label": "Given Name"}, definition=_char)
-    ind_fs.fields.get_or_create(name="middle_name", attrs={"label": "Middle Name"}, definition=_char)
-    ind_fs.fields.get_or_create(name="national_id_issuer", definition=_char)
-    ind_fs.fields.get_or_create(name="national_id_no", definition=_char)
-    ind_fs.fields.get_or_create(name="national_id_photo", definition=_char)
-    ind_fs.fields.get_or_create(name="phone_no", definition=_char)
-    ind_fs.fields.get_or_create(name="primary_collector_id", attrs={"label": "Primary Collector for"}, definition=_char)
-    ind_fs.fields.get_or_create(
-        name="relationship", attrs={"label": "Relationship", "required": True}, definition=_i_relationship
-    )
-    ind_fs.fields.get_or_create(name="role", attrs={"label": "Role"}, definition=_i_role)
+    people_fields_spec: list[FieldSpec] = [
+        ("type", defs["p_type"], {"label": "People Type", "required": True}),
+        ("full_name", defs["char"], {"label": "Full Name", "required": True}),
+        ("country", defs["h_country"], {"label": "Country", "required": True}),
+        ("residence_status", defs["h_residence"], {"label": "Residence Status", "required": True}),
+        ("gender", defs["i_gender"], None),
+        ("birth_date", defs["date"], {"label": "Birth Date", "required": True}),
+    ]
 
-    pp_fs, __ = Fieldset.objects.get_or_create(name=PEOPLE_CHECKER_NAME)
-    pp_fs.fields.get_or_create(name="type", attrs={"label": "People Type", "required": True}, definition=_p_type)
-    pp_fs.fields.get_or_create(name="full_name", attrs={"label": "Full Name", "required": True}, definition=_char)
-    pp_fs.fields.get_or_create(name="country", attrs={"label": "Country", "required": True}, definition=_h_country)
-    pp_fs.fields.get_or_create(
-        name="residence_status", attrs={"label": "Residence Status", "required": True}, definition=_h_residence
-    )
-    pp_fs.fields.get_or_create(name="gender", definition=_i_gender)
-    pp_fs.fields.get_or_create(name="birth_date", attrs={"label": "Birth Date", "required": True}, definition=_date)
+    def _add_fields(fieldset: Fieldset, fields_spec: list[FieldSpec]) -> None:
+        for name, definition, attrs in fields_spec:
+            fieldset.fields.get_or_create(name=name, definition=definition, defaults={"attrs": attrs or {}})
 
-    hh_dc, __ = DataChecker.objects.get_or_create(name=HOUSEHOLD_CHECKER_NAME)
-    hh_dc.fieldsets.add(hh_fs)
-    ind_dc, __ = DataChecker.objects.get_or_create(name=INDIVIDUAL_CHECKER_NAME)
-    ind_dc.fieldsets.add(ind_fs)
-    pp_dc, __ = DataChecker.objects.get_or_create(name=PEOPLE_CHECKER_NAME)
-    pp_dc.fieldsets.add(pp_fs)
+    hh_fs, _ = Fieldset.objects.get_or_create(name=HOUSEHOLD_CHECKER_NAME)
+    ind_fs, _ = Fieldset.objects.get_or_create(name=INDIVIDUAL_CHECKER_NAME)
+    pp_fs, _ = Fieldset.objects.get_or_create(name=PEOPLE_CHECKER_NAME)
+
+    _add_fields(hh_fs, household_fields_spec)
+    _add_fields(ind_fs, individual_fields_spec)
+    _add_fields(pp_fs, people_fields_spec)
+
+    hh_dc, _ = DataChecker.objects.get_or_create(name=HOUSEHOLD_CHECKER_NAME)
+    ind_dc, _ = DataChecker.objects.get_or_create(name=INDIVIDUAL_CHECKER_NAME)
+    pp_dc, _ = DataChecker.objects.get_or_create(name=PEOPLE_CHECKER_NAME)
+
+    hh_dc.fieldsets.set([hh_fs])
+    ind_dc.fieldsets.set([ind_fs])
+    pp_dc.fieldsets.set([pp_fs])
 
 
 def removes_hope_checkers() -> None:
     DataChecker.objects.filter(name=HOUSEHOLD_CHECKER_NAME).delete()
     DataChecker.objects.filter(name=INDIVIDUAL_CHECKER_NAME).delete()
+    DataChecker.objects.filter(name=PEOPLE_CHECKER_NAME).delete()
     Fieldset.objects.filter(name=HOUSEHOLD_CHECKER_NAME).delete()
     Fieldset.objects.filter(name=INDIVIDUAL_CHECKER_NAME).delete()
     Fieldset.objects.filter(name=PEOPLE_CHECKER_NAME).delete()
-    Fieldset.objects.filter(name=PEOPLE_CHECKER_NAME).delete()

tests/contrib/hope/test_hope_client.py

@@ -1,5 +1,6 @@
 import re
 from collections.abc import Callable
+from typing import Any
 from unittest.mock import Mock
 
 import pytest
@@ -182,33 +183,35 @@ def test_post_success(mocked_responses: responses.RequestsMock, mock_signals):
 
 
 @pytest.mark.parametrize(
-    ("status_code", "body", "expected_error"),
+    ("status_code", "body", "expected_error_pattern"),
     [
-        (400, {"error": "Bad request"}, "Error posting to https://hope-dummy.org/api/rest/dummy_path/:"),
-        (500, {"error": "Server error"}, "Error posting to https://hope-dummy.org/api/rest/dummy_path/:"),
-        (200, "invalid json", "Wrong JSON response posting to https://hope-dummy.org/api/rest/dummy_path/"),
+        pytest.param(400, {"error": "Bad request"}, r"HTTP error posting to.*?Status Code: 400", id="http_400"),
+        pytest.param(500, {"error": "Server error"}, r"HTTP error posting to.*?Status Code: 500", id="http_500"),
+        pytest.param(200, "invalid json", r"Wrong JSON response posting to .*?\. Status: 200", id="json_decode_error"),
     ],
 )
 @override_config(HOPE_API_URL="https://hope-dummy.org/api/rest", HOPE_API_TOKEN="dummy_token")
+@pytest.mark.django_db
 def test_post_errors(
     mocked_responses: responses.RequestsMock,
-    mock_signals,
+    mock_signals: Any,
     status_code: int,
-    body: dict | str,
-    expected_error: str,
-):
+    body: dict[str, Any] | str,
+    expected_error_pattern: str,
+) -> None:
     start_mock, end_mock = mock_signals
     client = HopeClient()
     path = "dummy_path"
     url = client.get_url(path)
     data = {"key": "value"}
 
     if isinstance(body, dict):
-        mocked_responses.add(responses.POST, url, json=body, status=status_code)
+        mocked_responses.add(method=responses.POST, url=url, json=body, status=status_code)
     else:
-        mocked_responses.add(responses.POST, url, body=body, status=status_code)
+        mocked_responses.add(method=responses.POST, url=url, body=body, status=status_code)
 
-    with pytest.raises(RemoteError, match=re.escape(expected_error)):
+    with pytest.raises(RemoteError, match=expected_error_pattern):
         client.post(path, data=data)
-    assert start_mock.call_count == 1
-    assert end_mock.call_count == 0
+
+    start_mock.assert_called_once()
+    end_mock.assert_not_called()