[Issue #6843] Replace encode-decode instances with utility function (#…

…6844)

* [Fixes #6797] replace encode-decode instances with utility function

* [#6797] Fix some spaces

* Added test function for surrogate_escape_string/2

* Remove misleading header

* flake8 fix

Co-authored-by: Matthew Northcott <[email protected]>
Co-authored-by: Alessio Fabiani <[email protected]>

geonode/layers/tests.py

@@ -53,7 +53,8 @@
     layer_type,
     get_files,
     get_valid_name,
-    get_valid_layer_name)
+    get_valid_layer_name,
+    surrogate_escape_string)
 from geonode.people.utils import get_valid_user
 from geonode.base.populate_test_data import all_public
 from geonode.base.models import TopicCategory, License, Region, Link
@@ -1056,6 +1057,17 @@ def test_batch_permissions(self):
             logger.debug(f" -- perm_spec[users] --> {perm_spec['users']}")
             self.assertTrue(user in perm_spec["users"])
 
+    def test_surrogate_escape_string(self):
+        surrogate_escape_raw = "Zo\udcc3\udcab"
+        surrogate_escape_expected = "Zoë"
+        surrogate_escape_result = surrogate_escape_string(
+            surrogate_escape_raw, 'UTF-8')  # add more test cases using different charsets?
+        self.assertEqual(
+            surrogate_escape_result,
+            surrogate_escape_expected,
+            "layers.utils.surrogate_escape_string did not produce expected result. "
+            f"Expected {surrogate_escape_expected}, received {surrogate_escape_result}")
+
 
 class UnpublishedObjectTests(GeoNodeBaseTestSupport):
 

geonode/layers/utils.py

@@ -1222,6 +1222,15 @@ def delete_orphaned_layers():
     return deleted
 
 
+def surrogate_escape_string(input_string, source_character_set):
+    """
+    Escapes a given input string using the provided source character set,
+    using the `surrogateescape` codec error handler.
+    """
+
+    return input_string.encode(source_character_set, "surrogateescape").decode("utf-8", "surrogateescape")
+
+
 def set_layers_permissions(permissions_name, resources_names=None,
                            users_usernames=None, groups_names=None,
                            delete_flag=None, verbose=False):

geonode/layers/views.py

@@ -82,7 +82,8 @@
 from geonode.layers.utils import (
     file_upload,
     is_raster,
-    is_vector)
+    is_vector,
+    surrogate_escape_string)
 
 from geonode.maps.models import Map
 from geonode.services.models import Service
@@ -380,7 +381,7 @@ def layer_upload(request, template='upload/layer_upload.html'):
         for _k in _keys:
             if _k in out:
                 if isinstance(out[_k], string_types):
-                    out[_k] = out[_k].encode(layer_charset, 'surrogateescape').decode('utf-8', 'surrogateescape')
+                    out[_k] = surrogate_escape_string(out[_k], layer_charset)
                 elif isinstance(out[_k], dict):
                     for key, value in out[_k].items():
                         try:
@@ -390,10 +391,8 @@ def layer_upload(request, template='upload/layer_upload.html'):
                                 out[_k][key] = item.as_text().encode(
                                     layer_charset, 'surrogateescape').decode('utf-8', 'surrogateescape')
                             else:
-                                out[_k][key] = item.encode(layer_charset, 'surrogateescape').decode(
-                                    'utf-8', 'surrogateescape')
-                            out[_k][key.encode(layer_charset, 'surrogateescape').decode(
-                                'utf-8', 'surrogateescape')] = out[_k].pop(key)
+                                out[_k][key] = surrogate_escape_string(item, layer_charset)
+                            out[_k][surrogate_escape_string(key, layer_charset)] = out[_k].pop(key)
                         except Exception as e:
                             logger.exception(e)