geocoding: allow custom url for search and reverse geocoding (#22218)

tests/test_api.py

@@ -2038,15 +2038,20 @@ def test_reverse_geocoding(pub):
         get_app(pub).get('/api/reverse-geocoding', status=400)
         resp = get_app(pub).get('/api/reverse-geocoding?lat=0&lon=0')
         assert resp.body == 'xxx'
-        assert urlopen.call_args[0][0] == 'http://nominatim.openstreetmap.org/reverse?format=json&zoom=18&addressdetails=1&lat=0&lon=0&accept-language=en'
+        assert urlopen.call_args[0][0] == 'http://nominatim.openstreetmap.org/reverse?zoom=18&format=json&addressdetails=1&lat=0&lon=0&accept-language=en'
 
         pub.site_options.add_section('options')
         pub.site_options.set('options', 'nominatim_reverse_zoom_level', '16')
         pub.site_options.write(open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w'))
         resp = get_app(pub).get('/api/reverse-geocoding?lat=0&lon=0')
-        assert urlopen.call_args[0][0] == 'http://nominatim.openstreetmap.org/reverse?format=json&zoom=16&addressdetails=1&lat=0&lon=0&accept-language=en'
+        assert urlopen.call_args[0][0] == 'http://nominatim.openstreetmap.org/reverse?zoom=16&format=json&addressdetails=1&lat=0&lon=0&accept-language=en'
 
         pub.site_options.set('options', 'nominatim_key', 'KEY')
         pub.site_options.write(open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w'))
         resp = get_app(pub).get('/api/reverse-geocoding?lat=0&lon=0')
-        assert urlopen.call_args[0][0] == 'http://nominatim.openstreetmap.org/reverse?format=json&zoom=16&addressdetails=1&lat=0&lon=0&accept-language=en&key=KEY'
+        assert urlopen.call_args[0][0] == 'http://nominatim.openstreetmap.org/reverse?zoom=16&key=KEY&format=json&addressdetails=1&lat=0&lon=0&accept-language=en'
+
+        pub.site_options.set('options', 'reverse_geocoding_service_url', 'http://reverse.example.net/?param=value')
+        pub.site_options.write(open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w'))
+        resp = get_app(pub).get('/api/reverse-geocoding?lat=0&lon=0')
+        assert urlopen.call_args[0][0] == 'http://reverse.example.net/?param=value&format=json&addressdetails=1&lat=0&lon=0&accept-language=en'

tests/test_workflows.py

@@ -2328,10 +2328,38 @@ def test_geolocate_address(pub):
         http_get_page.return_value = (None, 200,
                 json.dumps([{'lat':'48.8337085','lon':'2.3233693'}]), None)
         item.perform(formdata)
+        assert 'http://nominatim.openstreetmap.org/search' in http_get_page.call_args[0][0]
         assert urllib2.quote('169 rue du chateau, paris') in http_get_page.call_args[0][0]
         assert int(formdata.geolocations['base']['lat']) == 48
         assert int(formdata.geolocations['base']['lon']) == 2
 
+    pub.load_site_options()
+    pub.site_options.set('options', 'nominatim_key', 'KEY')
+    with mock.patch('wcs.wf.geolocate.http_get_page') as http_get_page:
+        http_get_page.return_value = (None, 200,
+                json.dumps([{'lat':'48.8337085', 'lon':'2.3233693'}]), None)
+        item.perform(formdata)
+        assert 'http://nominatim.openstreetmap.org/search' in http_get_page.call_args[0][0]
+        assert urllib2.quote('169 rue du chateau, paris') in http_get_page.call_args[0][0]
+        assert 'key=KEY' in http_get_page.call_args[0][0]
+        assert int(formdata.geolocations['base']['lat']) == 48
+        assert int(formdata.geolocations['base']['lon']) == 2
+
+    pub.load_site_options()
+    pub.site_options.set('options', 'geocoding_service_url', 'http://example.net/')
+    with mock.patch('wcs.wf.geolocate.http_get_page') as http_get_page:
+        http_get_page.return_value = (None, 200,
+                json.dumps([{'lat':'48.8337085','lon':'2.3233693'}]), None)
+        item.perform(formdata)
+        assert 'http://example.net/?q=' in http_get_page.call_args[0][0]
+
+    pub.site_options.set('options', 'geocoding_service_url', 'http://example.net/?param=value')
+    with mock.patch('wcs.wf.geolocate.http_get_page') as http_get_page:
+        http_get_page.return_value = (None, 200,
+                json.dumps([{'lat':'48.8337085','lon':'2.3233693'}]), None)
+        item.perform(formdata)
+        assert 'http://example.net/?param=value&' in http_get_page.call_args[0][0]
+
     # check for invalid ezt
     item.address_string = '[if-any], paris, france'
     formdata.geolocations = None

wcs/api.py

@@ -683,17 +683,13 @@ class ApiDirectory(Directory):
             lon = get_request().form['lon']
         except KeyError:
             raise QueryError
-        nominatim_url = get_publisher().get_site_option('nominatim_url')
-        if not nominatim_url:
-            nominatim_url = 'http://nominatim.openstreetmap.org'
-        nominatim_reverse_zoom_level = get_publisher().get_site_option(
-                'nominatim_reverse_zoom_level') or 18
-        get_response().set_content_type('application/json')
-        url = '%s/reverse?format=json&zoom=%s&addressdetails=1&lat=%s&lon=%s' % (
-                nominatim_url, nominatim_reverse_zoom_level, lat, lon)
+        url = get_publisher().get_reverse_geocoding_service_url()
+        if '?' in url:
+            url += '&'
+        else:
+            url += '?'
+        url += 'format=json&addressdetails=1&lat=%s&lon=%s' % (lat, lon)
         url += '&accept-language=%s' % (get_publisher().get_site_language() or 'en')
-        if get_publisher().get_site_option('nominatim_key'):
-            url += '&key=' + get_publisher().get_site_option('nominatim_key')
         return misc.urlopen(url).read()
 
     def roles(self):

wcs/qommon/publisher.py

@@ -951,6 +951,30 @@ class QommonPublisher(Publisher, object):
                 'https://{s}.tile.openstreetmap.org/{z}/{x}/{y}.png'
         return attrs
 
+    def get_reverse_geocoding_service_url(self):
+        url = self.get_site_option('reverse_geocoding_service_url')
+        if url:
+            return url
+        url = self.get_site_option('nominatim_url') or 'http://nominatim.openstreetmap.org'
+        url += '/reverse'
+        reverse_zoom_level = self.get_site_option('nominatim_reverse_zoom_level') or 18
+        url += '?zoom=%s' % reverse_zoom_level
+        key = self.get_site_option('nominatim_key')
+        if key:
+            url += '&key=%s' % key
+        return url
+
+    def get_geocoding_service_url(self):
+        url = self.get_site_option('geocoding_service_url')
+        if url:
+            return url
+        url = self.get_site_option('nominatim_url') or 'http://nominatim.openstreetmap.org'
+        url += '/search'
+        key = self.get_site_option('nominatim_key')
+        if key:
+            url += '?key=%s' % key
+        return url
+
     def get_supported_authentication_contexts(self):
         contexts = collections.OrderedDict()
         labels = {

wcs/wf/geolocate.py

@@ -103,10 +103,6 @@ class GeolocateWorkflowStatusItem(WorkflowStatusItem):
             formdata.store()
 
     def geolocate_address_string(self, formdata):
-        nominatim_url = get_publisher().get_site_option('nominatim_url')
-        if not nominatim_url:
-            nominatim_url = 'http://nominatim.openstreetmap.org'
-
         try:
             address = self.compute(self.address_string, raises=True)
         except Exception, e:
@@ -116,11 +112,14 @@ class GeolocateWorkflowStatusItem(WorkflowStatusItem):
         if not address:
             get_logger().debug('error geolocating string (empty string)')
             return
-
-        url = '%s/search?q=%s&format=json' % (nominatim_url, urllib2.quote(address))
+        url = get_publisher().get_geocoding_service_url()
+        if '?' in url:
+            url += '&'
+        else:
+            url += '?'
+        url += 'q=%s' % urllib2.quote(address)
+        url += '&format=json'
         url += '&accept-language=%s' % (get_publisher().get_site_language() or 'en')
-        if get_publisher().get_site_option('nominatim_key'):
-            url += '&key=' + get_publisher().get_site_option('nominatim_key')
 
         response, status, data, auth_header = http_get_page(url)
         if status != 200: