Merge pull request #400 from tilezen/travisg/20211005-fix-hstore-esca…

…ping

add comma as character to requiring hstore escape

tests/test_wof.py

@@ -1,5 +1,7 @@
 import unittest
 
+from tilequeue.wof import escape_hstore_string
+
 
 class TestNeighbourhoodDiff(unittest.TestCase):
 
@@ -294,3 +296,30 @@ def test_float_min_max_zoom(self):
         neighbourhood = self._call_fut(14.2, 16.8)
         self.assertEqual(neighbourhood.min_zoom, 14.2)
         self.assertEqual(neighbourhood.max_zoom, 16.8)
+
+class TestEscapeHStoreString(unittest.TestCase):
+
+    def test_has_spaces(self):
+        test = "a b c"
+        expected = "\"a b c\""
+        self.assertEqual(expected, escape_hstore_string(test))
+
+    def test_has_commas(self):
+        test = "a,b"
+        expected = "\"a,b\""
+        self.assertEqual(expected, escape_hstore_string(test))
+
+    def test_has_quote(self):
+        test = 'a"b '
+        expected = '\"a\\\\"b \"'
+        self.assertEqual(expected, escape_hstore_string(test))
+
+    def test_nothing_to_escape(self):
+        test = "normalstring"
+        expected = test
+        self.assertEqual(expected, escape_hstore_string(test))
+
+    def test_escape_for_several_reasons(self):
+        test = 'semicolons and, "oxford commas" are cool'
+        expected = '"semicolons and, \\\\"oxford commas\\\\" are cool"'
+        self.assertEqual(expected, escape_hstore_string(test))

tilequeue/wof.py

@@ -600,23 +600,25 @@ def create_neighbourhood_file_object(neighbourhoods, curdate=None):
     return buf
 
 
+def escape_string(s):
+    return s.encode('utf-8').replace('\t', ' ').replace('\n', ' ')
+
+
+def escape_hstore_string(s):
+    s = escape_string(s)
+    if ' ' in s or ',' in s:
+        s = s.replace('"', '\\\\"')
+        s = '"%s"' % s
+    return s
+
+
 def write_neighbourhood_data_to_file(buf, neighbourhoods, curdate=None):
     if curdate is None:
         curdate = datetime.now().date()
 
     # tell shapely to include the srid when generating WKBs
     geos.WKBWriter.defaults['include_srid'] = True
 
-    def escape_string(s):
-        return s.encode('utf-8').replace('\t', ' ').replace('\n', ' ')
-
-    def escape_hstore_string(s):
-        s = escape_string(s)
-        if ' ' in s:
-            s = s.replace('"', '\\\\"')
-            s = '"%s"' % s
-        return s
-
     def write_nullable_int(buf, x):
         if x is None:
             buf.write('\\N\t')