misc: do not compute variadic URLs using complex variables (#55578)

tests/test_datasource.py

@@ -9,8 +9,10 @@ import pytest
 
 from wcs import data_sources, fields
 from wcs.data_sources import NamedDataSource, register_data_source_function
+from wcs.formdef import FormDef
 from wcs.qommon.form import Form, get_request
 from wcs.qommon.http_request import HTTPRequest
+from wcs.workflows import WorkflowStatusItem
 
 from .test_widgets import MockHtmlForm, mock_form_submission
 from .utilities import clean_temporary_pub, create_temporary_pub
@@ -1200,8 +1202,6 @@ def test_named_datasource_id_parameter(requests_pub):
 
 
 def test_named_datasource_in_formdef(pub):
-    from wcs.formdef import FormDef
-
     NamedDataSource.wipe()
     datasource = NamedDataSource(name='foobar')
     datasource.data_source = {'type': 'json', 'value': 'http://whatever/'}
@@ -1218,3 +1218,38 @@ def test_named_datasource_in_formdef(pub):
 
     datasource.slug = 'barfoo'
     assert not datasource.is_used_in_formdef(formdef)
+
+
+def test_data_source_in_template(pub):
+    NamedDataSource.wipe()
+
+    FormDef.wipe()
+    formdef = FormDef()
+    formdef.name = 'foobar'
+    formdef.workflow_options = {'foo': 'hello'}
+    formdef.fields = [
+        fields.StringField(id='1', label='Test', type='string', varname='foo'),
+    ]
+    formdef.store()
+
+    formdata = formdef.data_class()()
+    formdata.data = {'1': 'hello'}
+    formdata.store()
+    pub.substitutions.feed(formdef)
+    pub.substitutions.feed(formdata)
+
+    data_source = NamedDataSource(name='foobar')
+    data_source.data_source = {'type': 'json', 'value': "https://example.invalid/json?t={{form_var_foo}}"}
+    data_source.store()
+
+    with pub.complex_data():
+        with mock.patch('wcs.qommon.misc.urlopen') as urlopen:
+            urlopen.side_effect = lambda *args: io.StringIO(
+                '{"data": [{"id": 0, "text": "zéro"}, {"id": 1, "text": "uné"}, {"id": 2, "text": "deux"}]}'
+            )
+
+            assert (
+                WorkflowStatusItem.compute('{{ data_source.foobar|first|get:"text" }}', allow_complex=True)
+                == 'zéro'
+            )
+            assert urlopen.call_args[0][0] == 'https://example.invalid/json?t=hello'

wcs/qommon/misc.py

@@ -29,6 +29,7 @@ import time
 import unicodedata
 import urllib.parse
 import xml.etree.ElementTree as ET
+from contextlib import contextmanager
 
 import requests
 from requests.adapters import HTTPAdapter
@@ -409,6 +410,16 @@ def http_delete_request(url, **kwargs):
     return _http_request(url, 'DELETE', **kwargs)
 
 
+@contextmanager
+def no_complex(context):
+    allow_complex = context.get('allow_complex')
+    context['allow_complex'] = False
+    try:
+        yield context
+    finally:
+        context['allow_complex'] = allow_complex
+
+
 def get_variadic_url(url, variables, encode_query=True):
     if not Template.is_template_string(url):
         return url
@@ -416,7 +427,8 @@ def get_variadic_url(url, variables, encode_query=True):
     # django template
     if '{{' in url or '{%' in url:
         try:
-            url = Template(url).render(variables)
+            with no_complex(variables):
+                url = Template(url).render(variables)
             p = urllib.parse.urlsplit(url)
             scheme, netloc, path, query, fragment = (p.scheme, p.netloc, p.path, p.query, p.fragment)
             if path.startswith('//'):