forms: allow form variables in geojson URLs (#51558)

tests/form_pages/test_all.py

@@ -8685,6 +8685,42 @@ def test_form_item_map_data_source(pub, http_requests):
     assert formdata.data['1_structured']['geometry']['coordinates'] == [1, 2]
 
 
+def test_form_item_dynamic_map_data_source(pub, http_requests):
+    NamedDataSource.wipe()
+    data_source = NamedDataSource(name='foobar')
+    data_source.data_source = {
+        'type': 'geojson',
+        'value': 'http://remote.example.net/geojson?{{ form_var_test }}',
+    }
+    data_source.id_property = 'id'
+    data_source.label_template_property = '{{ text }}'
+    data_source.cache_duration = '5'
+    data_source.store()
+
+    formdef = create_formdef()
+    formdef.fields = [
+        fields.PageField(id='0', label='1st page', type='page'),
+        fields.StringField(id='1', label='string', varname='test'),
+        fields.PageField(id='2', label='2nd page', type='page'),
+        fields.ItemField(id='3', label='map', display_mode='map', data_source={'type': 'foobar'}),
+    ]
+    formdef.store()
+    formdef.data_class().wipe()
+    app = get_app(pub)
+    resp = app.get('/test/')
+    resp.form['f1'] = 'plop'
+    resp = resp.form.submit('submit')  # -> 2nd page
+    markers_url = resp.pyquery('div[data-markers-radio-name]')[0].attrib['data-markers-url']
+    assert markers_url.startswith('/api/geojson/')
+    resp_geojson = app.get(markers_url)
+    assert len(resp_geojson.json['features']) == 2
+    assert http_requests.count() == 1
+    assert http_requests.get_last('url') == 'http://remote.example.net/geojson?plop'
+    resp_geojson = app.get(markers_url)
+    assert http_requests.count() == 1  # cache was used
+    assert len(resp_geojson.json['features']) == 2
+
+
 def test_form_item_timetable_data_source(pub, http_requests):
     NamedDataSource.wipe()
     data_source = NamedDataSource(name='foobar')

wcs/api.py

@@ -1039,12 +1039,20 @@ class AutocompleteDirectory(Directory):
 
 class GeoJsonDirectory(Directory):
     def _q_lookup(self, component):
+        url = None
         try:
             data_source = get_data_source_object({'type': component}, ignore_errors=False)
         except KeyError:
-            raise TraversalError()
+            info = get_session().get_data_source_query_info_from_token(component)
+            if not info:
+                raise TraversalError()
+            try:
+                data_source = get_data_source_object({'type': info['slug']}, ignore_errors=False)
+            except KeyError:
+                raise TraversalError()
+            url = info['url']
         get_response().set_content_type('application/json')
-        return json.dumps(data_source.get_geojson_data())
+        return json.dumps(data_source.get_geojson_data(force_url=url))
 
 
 class ApiDirectory(Directory):

wcs/data_sources.py

@@ -561,13 +561,23 @@ class NamedDataSource(XmlStorableObject):
 
     def get_geojson_url(self):
         assert self.type == 'geojson'
-        return '/api/geojson/%s' % self.slug
-
-    def get_geojson_data(self):
         url = self.data_source.get('value').strip()
         if Template.is_template_string(url):
-            vars = get_publisher().substitutions.get_context_variables(mode='lazy')
-            url = get_variadic_url(url, vars)
+            context = get_publisher().substitutions.get_context_variables(mode='lazy')
+            new_url = get_variadic_url(url, context)
+            if new_url != url:
+                info = {'url': new_url, 'slug': self.slug}
+                return '/api/geojson/%s' % get_session().get_data_source_query_info_token(info)
+        return '/api/geojson/%s' % self.slug
+
+    def get_geojson_data(self, force_url=None):
+        if force_url:
+            url = force_url
+        else:
+            url = self.data_source.get('value').strip()
+            if Template.is_template_string(url):
+                context = get_publisher().substitutions.get_context_variables(mode='lazy')
+                url = get_variadic_url(url, context)
 
         request = get_request()
         if hasattr(request, 'datasources_cache') and url in request.datasources_cache: