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

Frédéric Péters, 09 octobre 2016 20:37

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_api.py        | 49 ++++++++++++++++++++++++++++++++++++++++++++++++
 tests/test_form_pages.py | 12 +++++++++++-
 wcs/api.py               | 42 +++++++++++++++++++++++++++++++++++++++--
 wcs/data_sources.py      | 12 ++++++++----
 wcs/fields.py            | 19 ++++++++++++++++++-
 wcs/qommon/form.py       |  3 +--
 6 files changed, 127 insertions(+), 10 deletions(-)

     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)

     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 qommon.tokens import Token
     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):
             try:
                 token = Token.get(component)
             except KeyError:
                 raise AccessForbiddenError()
             if not getattr(token, 'session_id', None):
                 raise AccessForbiddenError()
             if token.session_id != get_session().session_id:
                 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 *
     from qommon.misc import localstrftime, date_format, ellipsize
     from qommon.tokens import Token
     from qommon import get_cfg, get_logger
     from qommon.strftime import strftime
-...
                 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 = Token()
                     token.session_id = get_session().session_id
                     token.store()
                     kwargs['url'] = get_publisher().get_root_url() + 'api/datasources/%s/%s' % (
                             self.data_source.get('type'), token.id)
                     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