Show field names in bulk upload error file (#2534)

local_units/bulk_upload.py

@@ -123,19 +123,30 @@ def __init__(self, fieldnames: list[str], header_map: dict[str, str]):
         self._has_errors = False
 
     def _format_errors(self, errors: dict) -> dict[str, list[str]]:
-        """Recursively flatten DRF errors."""
+        """Recursively flatten DRF errors onto spreadsheet column labels."""
         formatted = {}
         for key, value in errors.items():
             if isinstance(value, dict):
                 formatted.update(self._format_errors(value))
             elif isinstance(value, list):
-                header = self._reverse_header_map.get(key, key)
-                formatted[header] = [self._clean_message(v) for v in value]
+                self._assign_error(formatted, key, [self._clean_message(v) for v in value])
             else:
-                header = self._reverse_header_map.get(key, key)
-                formatted[header] = [self._clean_message(value)]
+                self._assign_error(formatted, key, [self._clean_message(value)])
         return formatted
 
+    def _assign_error(self, formatted: dict[str, list[str]], key: str, messages: list[str]) -> None:
+        if key == "non_field_errors":
+            formatted["General Error"] = messages
+            return
+
+        if key == "location":
+            for header in ("Latitude", "Longitude"):
+                formatted[header] = messages
+            return
+
+        header = self._reverse_header_map.get(key, key)
+        formatted[header] = messages
+
     def _clean_message(self, msg: any) -> str:
         if isinstance(msg, ErrorDetail):
             return str(msg)

local_units/serializers.py

@@ -968,16 +968,23 @@ def validate_level(self, value):
     def validate(self, validated_data):
 
         if not validated_data.get("local_branch_name") and not validated_data.get("english_branch_name"):
-            raise serializers.ValidationError(gettext("Branch Name Combination is required."))
+            message = gettext("Branch Name Combination is required.")
+            raise serializers.ValidationError(
+                {
+                    "local_branch_name": message,
+                    "english_branch_name": message,
+                }
+            )
 
         # Country location validation
         latitude = validated_data.pop("latitude")
         longitude = validated_data.pop("longitude")
         if not latitude or not longitude:
-            raise serializers.ValidationError(gettext("Latitude and Longitude are required."))
+            message = gettext("Latitude and Longitude are required.")
+            raise serializers.ValidationError({"latitude": message, "longitude": message})
         country = validated_data.get("country")
         if not country:
-            raise serializers.ValidationError(gettext("Country is required."))
+            raise serializers.ValidationError({"non_field_errors": gettext("Country is required.")})
 
         input_point = Point(longitude, latitude)
         if country.bbox:

local_units/test_bulk_upload_errors.py

@@ -0,0 +1,57 @@
+import unittest
+from types import SimpleNamespace
+
+from rest_framework.exceptions import ErrorDetail
+
+from local_units.bulk_upload import ErrorWriter
+
+
+class ErrorWriterTests(unittest.TestCase):
+    HEADER_MAP = {
+        "Local Unit Name (En)": "english_branch_name",
+        "Latitude": "latitude",
+        "Longitude": "longitude",
+    }
+
+    def _writer(self):
+        return ErrorWriter(
+            fieldnames=["Local Unit Name (En)", "Latitude", "Longitude"],
+            header_map=self.HEADER_MAP,
+        )
+
+    def test_maps_serializer_field_to_display_header(self):
+        writer = self._writer()
+        formatted = writer._format_errors({"latitude": ["This field is required."]})
+        self.assertIn("Latitude", formatted)
+        self.assertEqual(formatted["Latitude"], ["This field is required."])
+
+    def test_maps_non_field_errors_to_general_error_column(self):
+        writer = self._writer()
+        formatted = writer._format_errors({"non_field_errors": ["Branch Name Combination is required."]})
+        self.assertIn("General Error", formatted)
+        self.assertNotIn("non_field_errors", formatted)
+
+    def test_maps_location_error_to_latitude_and_longitude(self):
+        writer = self._writer()
+        formatted = writer._format_errors({"location": ["Input coordinates is outside country boundary"]})
+        self.assertEqual(formatted["Latitude"], ["Input coordinates is outside country boundary"])
+        self.assertEqual(formatted["Longitude"], ["Input coordinates is outside country boundary"])
+
+    def test_write_failed_row_includes_error_columns(self):
+        writer = self._writer()
+        writer.write(
+            {"english_branch_name": "Test", "latitude": "", "longitude": ""},
+            status=SimpleNamespace(name="FAILED"),
+            error_detail={
+                "latitude": [ErrorDetail("Latitude and Longitude are required.", code="invalid")],
+                "longitude": [ErrorDetail("Latitude and Longitude are required.", code="invalid")],
+            },
+        )
+        headers = [writer._ws.cell(row=1, column=i).value for i in range(1, writer._ws.max_column + 1)]
+        self.assertIn("Latitude_error", headers)
+        self.assertIn("Longitude_error", headers)
+        self.assertEqual(writer._ws.cell(row=2, column=headers.index("Latitude_error") + 1).value, "Latitude and Longitude are required.")
+
+
+if __name__ == "__main__":
+    unittest.main()