From 1853ffedaac84657f316865cedd83f7fe1c435a9 Mon Sep 17 00:00:00 2001 From: Tarek Lokal Date: Sat, 11 Jul 2026 13:37:08 +0200 Subject: [PATCH] fix: show field names in bulk upload error file (#2534) Map validation errors to spreadsheet column labels: non_field_errors become General Error, location errors attach to Latitude/Longitude, and serializer validate() now raises field-keyed errors for branch names and coordinates. Co-authored-by: Cursor --- local_units/bulk_upload.py | 21 +++++++--- local_units/serializers.py | 13 ++++-- local_units/test_bulk_upload_errors.py | 57 ++++++++++++++++++++++++++ 3 files changed, 83 insertions(+), 8 deletions(-) create mode 100644 local_units/test_bulk_upload_errors.py diff --git a/local_units/bulk_upload.py b/local_units/bulk_upload.py index 8687b78df..2676017fb 100644 --- a/local_units/bulk_upload.py +++ b/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) diff --git a/local_units/serializers.py b/local_units/serializers.py index b63cfe550..b50fa03a2 100644 --- a/local_units/serializers.py +++ b/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: diff --git a/local_units/test_bulk_upload_errors.py b/local_units/test_bulk_upload_errors.py new file mode 100644 index 000000000..ee3896604 --- /dev/null +++ b/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()