tests/form_pages/test_all.py

@@ -5314,6 +5314,67 @@ def test_form_autocomplete_named_datasource_expired_token(pub):
     assert '/api/autocomplete/%s' % token.id not in resp.text
 
 
+@pytest.mark.parametrize('sign', ['without-signature', 'with-signature'])
+def test_form_autocomplete_named_datasource_cache_duration(pub, sign):
+    CardDef.wipe()
+    FormDef.wipe()
+    pub.token_class.wipe()
+
+    formdef = FormDef()
+    formdef.name = 'test'
+    formdef.fields = [
+        fields.StringField(
+            id='0', label='string', type='string', required=False, data_source={'type': 'foobar'}
+        )
+    ]
+    formdef.store()
+
+    url = 'http://remote.example.net/json_%s_%s' % (hashlib.sha1(pub.app_dir.encode()).hexdigest(), sign)
+
+    NamedDataSource.wipe()
+    data_source = NamedDataSource(name='foobar')
+    data_source.data_source = {'type': 'json', 'value': url}
+    data_source.query_parameter = 'q'
+    data_source.id_parameter = 'id'
+    data_source.cache_duration = '1200'
+    data_source.store()
+
+    if sign == 'with-signature':
+        with open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w') as fd:
+            fd.write(
+                '''\
+[wscall-secrets]
+remote.example.net = 1234
+    '''
+            )
+
+    app = get_app(pub)
+    resp = app.get('/test/')
+    assert pub.token_class.count() == 1
+    token = pub.token_class.select()[0]
+    assert '/api/autocomplete/%s' % token.id in resp.text
+
+    with responses.RequestsMock() as rsps:
+        data = {
+            'data': [
+                {'id': '1', 'text': 'hello', 'extra': 'foo'},
+                {'id': '2', 'text': 'world', 'extra': 'bar'},
+            ]
+        }
+        rsps.get(url, json=data)
+
+        resp = app.get('/api/autocomplete/%s?q=a' % token.id)
+        assert len(rsps.calls) == 1
+        assert len(resp.json['data']) == 2
+        resp = app.get('/api/autocomplete/%s?q=a' % token.id)
+        assert len(rsps.calls) == 1  # cached
+        assert len(resp.json['data']) == 2
+
+        resp = app.get('/api/autocomplete/%s?q=b' % token.id)
+        assert len(rsps.calls) == 2  # not cached
+        assert len(resp.json['data']) == 2
+
+
 def test_form_workflow_trigger(pub):
     user = create_user(pub)
 

wcs/api.py

@@ -29,7 +29,7 @@ from quixote.errors import MethodNotAllowedError, RequestError
 
 import wcs.qommon.storage as st
 from wcs.admin.settings import UserFieldsFormDef
-from wcs.api_utils import get_query_flag, get_user_from_api_query_string, is_url_signed, sign_url_auto_orig
+from wcs.api_utils import get_query_flag, get_user_from_api_query_string, is_url_signed
 from wcs.carddef import CardDef
 from wcs.categories import Category
 from wcs.conditions import Condition, ValidationError
@@ -1192,10 +1192,10 @@ class AutocompleteDirectory(Directory):
             cache_duration = 0
             if info.get('data_source'):
                 named_data_source = NamedDataSource.get(info['data_source'])
-                cache_duration = named_data_source.cache_duration
+                if named_data_source.cache_duration:
+                    cache_duration = int(named_data_source.cache_duration)
             url = info['url']
             url += urllib.parse.quote(get_request().form.get('q', ''))
-            url = sign_url_auto_orig(url)
             get_response().set_content_type('application/json')
             entries = request_json_items(
                 url,

wcs/data_sources.py

@@ -248,11 +248,6 @@ def get_id_by_option_text(data_source, text_value):
 def get_json_from_url(
     url, data_source=None, log_message_part='JSON data source', raise_request_error=False, cache_duration=0
 ):
-    url = sign_url_auto_orig(url)
-    data_source = data_source or {}
-    data_key = data_source.get('data_attribute') or 'data'
-    geojson = data_source.get('type') == 'geojson'
-    error_summary = None
 
     if cache_duration:
         cache_key = 'data-source-cache-%s' % force_str(hashlib.md5(force_bytes(url)).hexdigest())
@@ -260,6 +255,12 @@ def get_json_from_url(
         if entries is not None:
             return entries
 
+    url = sign_url_auto_orig(url)
+    data_source = data_source or {}
+    data_key = data_source.get('data_attribute') or 'data'
+    geojson = data_source.get('type') == 'geojson'
+    error_summary = None
+
     try:
         entries = misc.json_loads(misc.urlopen(url).read())
         if not isinstance(entries, dict):