0001-misc-add-a-jsonp-endpoint-for-datasources-use-it-for.patch

Frédéric Péters, 12 octobre 2016 17:18

Voir les différences: en ligne côte à côte

Subject: [PATCH] misc: add a jsonp endpoint for datasources, use it for
 autocomplete (#10990)

 tests/test_admin_pages.py |  2 +-
 tests/test_api.py         | 49 +++++++++++++++++++++++++++++++++++++++++++++++
 tests/test_form_pages.py  | 12 +++++++++++-
 wcs/api.py                | 38 ++++++++++++++++++++++++++++++++++--
 wcs/data_sources.py       | 12 ++++++++----
 wcs/fields.py             | 16 +++++++++++++++-
 wcs/qommon/form.py        |  3 +--
 7 files changed, 121 insertions(+), 11 deletions(-)

                 resp.body.index('<label for="form_data_source">Data Source</label>')
         # start filling the "data source" field
         resp.forms[0]['data_source$type'] = 'JSON URL'
         resp.forms[0]['data_source$type'] = 'JSONP URL'
         resp = resp.forms[0].submit('data_source$apply')
         # it should now appear before the additional parameters section

     import hmac
     import base64
     import hashlib
     import re
     import urllib
     import urlparse
     import datetime
-...
         secret, orig = get_secret_and_orig('https://api.example.com/endpoint/')
         assert secret == '1234'
         assert orig == 'example.net'
     def test_datasources_jsonp(pub):
         NamedDataSource.wipe()
         data_source = NamedDataSource(name='foobar')
         source = [{'id': '1', 'text': 'foo', 'more': 'XXX'},
                   {'id': '2', 'text': 'bar', 'more': 'YYY'}]
         data_source.data_source = {'type': 'formula', 'value': repr(source)}
         data_source.store()
         get_app(pub).get('/api/datasources/xxx', status=404)
         get_app(pub).get('/api/datasources/xxx/', status=404)
         get_app(pub).get('/api/datasources/foobar/', status=403)
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'test'
         formdef.fields = [
             fields.StringField(id='0', label='foobar0', varname='foobar0',
                 data_source={'type': 'foobar'}),
+        ]
         formdef.store()
         get_app(pub).get('/api/datasources/foobar/12122', status=403)
         app = get_app(pub)
         resp = app.get('/test/')
         url = re.findall(r"'(/api/datasou.*)'", resp.body)[0]
         resp = app.get(url)
         assert len(resp.json['data']) == 2
         resp = app.get(url + '?q=fo')
         resp_data = resp.body
         assert len(resp.json['data']) == 1
         resp = app.get(url + '?q=fo&callback=cb123')
         assert resp_data in resp.body
         assert resp.body.startswith('cb123(')
         # check accessing the URL from another session
         get_app(pub).get(url, status=403)
         app2 = get_app(pub)
         resp2 = app2.get('/test/')
         app2.get(url, status=403)
         # test custom handling of jsonp sources (redirect)
         data_source.data_source = {'type': 'jsonp', 'value': 'http://remote.example.net/json'}
         data_source.store()
         resp = app.get(url + '?q=fo&callback=cb123')
         assert resp.location == 'http://remote.example.net/json?q=fo&callback=cb123'

         assert not hasattr(formdata.data['0'], 'metadata')
         assert not '0_structured' in formdata.data
     def test_form_string_field_autocomplete(pub):
         formdef = create_formdef()
         formdef.fields = [fields.StringField(id='0', label='string', type='string', required=False)]
-...
         assert ').autocomplete({' in resp.body
         assert 'http://example.net' in resp.body
         # named data source
         NamedDataSource.wipe()
         data_source = NamedDataSource(name='foobar')
         data_source.data_source = {'type': 'formula', 'value': repr([])}
         data_source.store()
         formdef.fields[0].data_source = {'type': 'foobar', 'value': ''}
         formdef.store()
         resp = get_app(pub).get('/test/')
         assert ').autocomplete({' in resp.body
         assert '/api/datasources/foobar/' in resp.body
     def test_form_workflow_trigger(pub):
         user = create_user(pub)

     # You should have received a copy of the GNU General Public License
     # along with this program; if not, see <http://www.gnu.org/licenses/>.
     import hashlib
     import json
     import time
     import urllib2
     import sys
     from quixote import get_request, get_publisher, get_response
     from quixote import get_request, get_publisher, get_response, get_session, redirect
     from quixote.directory import Directory
     from qommon import misc
     from qommon.errors import (AccessForbiddenError, QueryError, TraversalError,
         UnknownNameIdAccessForbiddenError)
     from wcs.categories import Category
     from wcs.data_sources import NamedDataSource
     from wcs.formdef import FormDef
     from wcs.roles import Role, logged_users_role
     from wcs.forms.common import FormStatusPage
-...
             return json.dumps(data)
     class ApiDataSourceDirectory(Directory):
         def __init__(self, datasource):
             self.datasource = datasource
         def _q_lookup(self, component):
             if not get_session() or not get_session().session_id:
                 raise AccessForbiddenError()
             if component != hashlib.md5('%s:%s' % (self.datasource.slug, get_session().session_id)).hexdigest():
                 raise AccessForbiddenError()
             dtype = self.datasource.data_source.get('type')
             if dtype == 'jsonp':
                 # redirect to the source
                 url = self.datasource.data_source.get('value')
                 if not '?' in url:
                     url += '?'
                 url += get_request().get_query()
                 return redirect(url)
             query = get_request().form.get('q', '').lower()
             items = [x[-1] for x in self.datasource.get_items() if query in x[1].lower()]
             return misc.json_response({'data': items})
     class ApiDataSourcesDirectory(Directory):
         def _q_lookup(self, component):
             try:
                 return ApiDataSourceDirectory(NamedDataSource.get_by_slug(component))
             except KeyError:
                 raise TraversalError()
     class ApiDirectory(Directory):
         _q_exports = ['forms', 'roles', ('reverse-geocoding', 'reverse_geocoding'),
                 'formdefs', 'categories', 'user', 'users', 'code']
                 'formdefs', 'categories', 'user', 'users', 'code',
                 'datasources']
         forms = ApiFormsDirectory()
         formdefs = ApiFormdefsDirectory()
-...
         user = ApiUserDirectory()
         users = ApiUsersDirectory()
         code = ApiTrackingCodeDirectory()
         datasources = ApiDataSourcesDirectory()
         def reverse_geocoding(self):
             try:

     class DataSourceSelectionWidget(CompositeWidget):
         def __init__(self, name, value=None, allow_jsonp=True,
                 allow_named_sources=True, **kwargs):
                 allow_named_sources=True, require_jsonp=False, **kwargs):
             CompositeWidget.__init__(self, name, value, **kwargs)
             if not value:
                 value = {}
             options = [('none', _('None')),
                        ('formula', _('Python Expression')),
                        ('json', _('JSON URL'))]
             options = [('none', _('None'))]
             if not require_jsonp:
                 options.append(('formula', _('Python Expression')))
                 options.append(('json', _('JSON URL')))
             if allow_jsonp:
                 options.append(('jsonp', _('JSONP URL')))
             if allow_named_sources:
-...
         def get_substitution_variables(cls):
             return {'data_source': DataSourcesSubstitutionProxy()}
         def get_items(self):
             return get_items(self.data_source)
     class DataSourcesSubstitutionProxy(object):
         def __getattr__(self, attr):

     import random
     import re
     import base64
     import hashlib
     import xml.etree.ElementTree as ET
     import collections
     from HTMLParser import HTMLParser
     from quixote import get_request, get_publisher
     from quixote import get_request, get_publisher, get_session_manager
     from quixote.html import htmltext, TemplateIO
     from qommon.form import *
-...
                 if real_data_source.get('type') == 'jsonp':
                     kwargs['url'] = real_data_source.get('value')
                     self.widget_class = AutocompleteStringWidget
                 elif self.data_source.get('type') not in ('none', 'formula', 'json'):
                     # named data source, pass a token to assert the user is
                     # allowed to access the data.
                     session = get_session()
                     if not session.session_id:
                         # this require a session to exist
                         session.get_anonymous_key(generate=True)
                         get_session_manager().maintain_session(session)
                     token = hashlib.md5('%s:%s' % (self.data_source.get('type'), get_session().session_id)).hexdigest()
                     kwargs['url'] = get_publisher().get_root_url() + 'api/datasources/%s/%s' % (
                             self.data_source.get('type'), token)
                     self.widget_class = AutocompleteStringWidget
         def fill_admin_form(self, form):
             WidgetField.fill_admin_form(self, form)
-...
                     value=self.validation, advanced=(not self.validation))
             form.add(data_sources.DataSourceSelectionWidget, 'data_source',
                      value=self.data_source,
                      require_jsonp=True,
                      title=_('Data Source'),
                      hint=_('This will allow autocompletion from an external source.'),
                      advanced=is_datasource_advanced(self.data_source),

         url = None
         def __init__(self, *args, **kwargs):
             self.url = kwargs.pop('url', None)
             WcsExtraStringWidget.__init__(self, *args, **kwargs)
             if kwargs.get('url'):
                 self.url = kwargs.get('url')
         def render_content(self):
             get_response().add_javascript(['jquery.js', 'jquery-ui.js'])
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0001-misc-add-a-jsonp-endpoint-for-datasources-use-it-for.patch