plone / plone.app.widgets / 665

pending completion

Build # 665

Build Type

Pull #101

travis-ci

Committed by

thet

Commit Message

<a href="https://github.com/plone/plone.app.widgets/commit/<a class=hub.com/plone/plone.app.widgets/commit/8046d08d0b71245054d9b5157f3347ccab875a2a">8046d08d0<a href="https://github.com/plone/plone.app.widgets/commit/8046d08d0b71245054d9b5157f3347ccab875a2a">&quot;&gt;Revert &amp;quot;Revert &amp;quot;Change getVocabulary to work also on non IPloneSiteRoot contexts.&quot;&quot;

This reverts commit </a><a class="double-link" href="https://github.com/plone/plone.app.widgets/commit/<a class="double-link" href="https://github.com/plone/plone.app.widgets/commit/4e02c2097b0dea077b855b3a573773fcc69b3217">4e02c2097</a>">4e02c2097</a><a href="https://github.com/plone/plone.app.widgets/commit/8046d08d0b71245054d9b5157f3347ccab875a2a">.

Pull Request Pull Request #101: Change getVocabulary to work also on non IPloneSiteRoot contexts

Run Details

8 of 8 new or added lines in 1 file covered. (100.0%)

2653 of 3061 relevant lines covered (86.67%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.84

/plone/app/widgets/browser/vocabulary.py

# -*- coding: utf-8 -*-
from AccessControl import getSecurityManager
from Products.CMFCore.utils import getToolByName
from Products.CMFPlone.interfaces import IPloneSiteRoot
from Products.Five import BrowserView
from logging import getLogger
from plone.app.querystring import queryparser
from plone.app.widgets.interfaces import IFieldPermissionChecker
from plone.autoform.interfaces import WRITE_PERMISSIONS_KEY
from plone.supermodel.utils import mergedTaggedValueDict
from types import FunctionType
from zope.component import getUtility
from zope.component import queryAdapter
from zope.component import queryUtility
from zope.schema.interfaces import ICollection
from zope.schema.interfaces import IVocabularyFactory
from zope.security.interfaces import IPermission
import inspect
import itertools
import json

logger = getLogger(__name__)


_permissions = {
    'plone.app.vocabularies.Users': 'Modify portal content',
    'plone.app.vocabularies.Catalog': 'View',
    'plone.app.vocabularies.Keywords': 'Modify portal content',
    'plone.app.vocabularies.SyndicatableFeedItems': 'Modify portal content'
}


def _parseJSON(s):
    if isinstance(s, basestring):
        s = s.strip()
        if (s.startswith('{') and s.endswith('}')) or \
                (s.startswith('[') and s.endswith(']')):  # detect if json
            return json.loads(s)
    return s


_unsafe_metadata = ['Creator', 'listCreators', 'author_name', 'commentors']
_safe_callable_metadata = ['getURL', 'getPath']


class VocabLookupException(Exception):
    pass


class BaseVocabularyView(BrowserView):

    def __call__(self):
        """
        Accepts GET parameters of:
        name: Name of the vocabulary
        field: Name of the field the vocabulary is being retrieved for
        query: string or json object of criteria and options.
            json value consists of a structure:
                {
                    criteria: object,
                    sort_on: index,
                    sort_order: (asc|reversed)
                }
        attributes: comma seperated, or json object list
        batch: {
            page: 1-based page of results,
            size: size of paged results
        }
        """
        context = self.get_context()
        self.request.response.setHeader("Content-type", "application/json")

        try:
            vocabulary = self.get_vocabulary()
        except VocabLookupException, e:
            return json.dumps({'error': e.message})

        results_are_brains = False
        if hasattr(vocabulary, 'search_catalog'):
            query = self.parsed_query()
            results = vocabulary.search_catalog(query)
            results_are_brains = True
        elif hasattr(vocabulary, 'search'):
            try:
                query = self.parsed_query()['SearchableText']['query']
            except KeyError:
                results = iter(vocabulary)
            else:
                results = vocabulary.search(query)
        else:
            results = vocabulary

        try:
            total = len(results)
        except TypeError:
            total = 0  # do not error if object does not support __len__
                       # we'll check again later if we can figure some size
                       # out

        # get batch
        batch = _parseJSON(self.request.get('batch', ''))
        if batch and ('size' not in batch or 'page' not in batch):
            batch = None  # batching not providing correct options
        if batch:
            # must be slicable for batching support
            page = int(batch['page'])
            # page is being passed in is 1-based
            start = (max(page - 1, 0)) * int(batch['size'])
            end = start + int(batch['size'])
            # Try __getitem__-based slice, then iterator slice.
            # The iterator slice has to consume the iterator through
            # to the desired slice, but that shouldn't be the end
            # of the world because at some point the user will hopefully
            # give up scrolling and search instead.
            try:
                results = results[start:end]
            except TypeError:
                results = itertools.islice(results, start, end)

        # build result items
        items = []

        attributes = _parseJSON(self.request.get('attributes', ''))
        if isinstance(attributes, basestring) and attributes:
            attributes = attributes.split(',')

        if attributes:
            portal = getToolByName(context, 'portal_url').getPortalObject()
            base_path = '/'.join(portal.getPhysicalPath())
            for vocab_item in results:
                if not results_are_brains:
                    vocab_item = vocab_item.value
                item = {}
                for attr in attributes:
                    key = attr
                    if ':' in attr:
                        key, attr = attr.split(':', 1)
                    if attr in _unsafe_metadata:
                        continue
                    if key == 'path':
                        attr = 'getPath'
                    val = getattr(vocab_item, attr, None)
                    if callable(val):
                        if attr in _safe_callable_metadata:
                            val = val()
                        else:
                            continue
                    if key == 'path':
                        val = val[len(base_path):]
                    item[key] = val
                items.append(item)
        else:
            for item in results:
                items.append({'id': item.token, 'text': item.title})

        if total == 0:
            total = len(items)

        return json.dumps({
            'results': items,
            'total': total
        })

    def parsed_query(self, ):
        query = _parseJSON(self.request.get('query', '')) or {}
        if query:
            parsed = queryparser.parseFormquery(
                self.get_context(), query['criteria'])
            if 'sort_on' in query:
                parsed['sort_on'] = query['sort_on']
            if 'sort_order' in query:
                parsed['sort_order'] = str(query['sort_order'])
            query = parsed
        return query


class VocabularyView(BaseVocabularyView):
    """Queries a named vocabulary and returns JSON-formatted results."""

    def get_context(self):
        return self.context

    def get_vocabulary(self):
        # Look up named vocabulary and check permission.

        context = self.context
        factory_name = self.request.get('name', None)
        field_name = self.request.get('field', None)
        if not factory_name:
            raise VocabLookupException('No factory provided.')
        authorized = None
        sm = getSecurityManager()
        if factory_name in _permissions:
            # Short circuit if permission is in global registry
            authorized = sm.checkPermission(
                _permissions[factory_name], context
            )
        elif field_name:
            # Check field specific permission
            permission_checker = queryAdapter(
                context, IFieldPermissionChecker
            )
            if permission_checker is not None:
                authorized = permission_checker.validate(
                    field_name, factory_name
                )

        if not authorized:
            raise VocabLookupException('Vocabulary lookup not allowed')

        factory = queryUtility(IVocabularyFactory, factory_name)
        if not factory:
            raise VocabLookupException(
                'No factory with name "%s" exists.' % factory_name)

        # This part is for backwards-compatibility with the first
        # generation of vocabularies created for plone.app.widgets,
        # which take the (unparsed) query as a parameter of the vocab
        # factory rather than as a separate search method.
        if type(factory) is FunctionType:
            factory_spec = inspect.getargspec(factory)
        else:
            factory_spec = inspect.getargspec(factory.__call__)
        query = _parseJSON(self.request.get('query', ''))
        if query and 'query' in factory_spec.args:
            vocabulary = factory(context, query=query)
        else:
            # This is what is reached for non-legacy vocabularies.
            vocabulary = factory(context)

        return vocabulary


class SourceView(BaseVocabularyView):
    """Queries a field's source and returns JSON-formatted results."""

    def get_context(self):
        return self.context.context

    def get_vocabulary(self):
        widget = self.context
        field = widget.field.bind(widget.context)

        # check field's write permission
        info = mergedTaggedValueDict(field.interface, WRITE_PERMISSIONS_KEY)
        permission_name = info.get(field.__name__, 'cmf.ModifyPortalContent')
        permission = queryUtility(IPermission, name=permission_name)
        if permission is None:
            permission = getUtility(
                IPermission, name='cmf.ModifyPortalContent')
        if not getSecurityManager().checkPermission(
                permission.title, self.get_context()):
            raise VocabLookupException('Vocabulary lookup not allowed.')

        if ICollection.providedBy(field):
            return field.value_type.vocabulary
        else:
            return field.vocabulary

1	# -- coding: utf-8 --
2	from AccessControl import getSecurityManager	1✔
3	from Products.CMFCore.utils import getToolByName	1✔
4	from Products.CMFPlone.interfaces import IPloneSiteRoot	1✔
5	from Products.Five import BrowserView	1✔
6	from logging import getLogger	1✔
7	from plone.app.querystring import queryparser	1✔
8	from plone.app.widgets.interfaces import IFieldPermissionChecker	1✔
9	from plone.autoform.interfaces import WRITE_PERMISSIONS_KEY	1✔
10	from plone.supermodel.utils import mergedTaggedValueDict	1✔
11	from types import FunctionType	1✔
12	from zope.component import getUtility	1✔
13	from zope.component import queryAdapter	1✔
14	from zope.component import queryUtility	1✔
15	from zope.schema.interfaces import ICollection	1✔
16	from zope.schema.interfaces import IVocabularyFactory	1✔
17	from zope.security.interfaces import IPermission	1✔
18	import inspect	1✔
19	import itertools	1✔
20	import json	1✔
21
22	logger = getLogger(__name__)	1✔
23
24
25	_permissions = {	1✔
26	'plone.app.vocabularies.Users': 'Modify portal content',
27	'plone.app.vocabularies.Catalog': 'View',
28	'plone.app.vocabularies.Keywords': 'Modify portal content',
29	'plone.app.vocabularies.SyndicatableFeedItems': 'Modify portal content'
30	}
31
32
33	def _parseJSON(s):	1✔
34	if isinstance(s, basestring):	1✔
35	s = s.strip()	1✔
36	if (s.startswith('{') and s.endswith('}')) or \	1✔
37	(s.startswith('[') and s.endswith(']')): # detect if json
38	return json.loads(s)	1✔
39	return s	1✔
40
41
42	_unsafe_metadata = ['Creator', 'listCreators', 'author_name', 'commentors']	1✔
43	_safe_callable_metadata = ['getURL', 'getPath']	1✔
44
45
46	class VocabLookupException(Exception):	1✔
47	pass	1✔
48
49
50	class BaseVocabularyView(BrowserView):	1✔
51
52	def __call__(self):	1✔
53	"""
54	Accepts GET parameters of:
55	name: Name of the vocabulary
56	field: Name of the field the vocabulary is being retrieved for
57	query: string or json object of criteria and options.
58	json value consists of a structure:
59	{
60	criteria: object,
61	sort_on: index,
62	sort_order: (asc\|reversed)
63	}
64	attributes: comma seperated, or json object list
65	batch: {
66	page: 1-based page of results,
67	size: size of paged results
68	}
69	"""
70	context = self.get_context()	1✔
71	self.request.response.setHeader("Content-type", "application/json")	1✔
72
73	try:	1✔
74	vocabulary = self.get_vocabulary()	1✔
75	except VocabLookupException, e:	1✔
76	return json.dumps({'error': e.message})	1✔
77
78	results_are_brains = False	1✔
79	if hasattr(vocabulary, 'search_catalog'):	1✔
80	query = self.parsed_query()	1✔
81	results = vocabulary.search_catalog(query)	1✔
82	results_are_brains = True	1✔
83	elif hasattr(vocabulary, 'search'):	1✔
84	try:	1✔
85	query = self.parsed_query()['SearchableText']['query']	1✔
86	except KeyError:	×
87	results = iter(vocabulary)	×
88	else:
89	results = vocabulary.search(query)	1✔
90	else:
91	results = vocabulary	1✔
92
93	try:	1✔
94	total = len(results)	1✔
95	except TypeError:	1✔
96	total = 0 # do not error if object does not support __len__	1✔
97	# we'll check again later if we can figure some size
98	# out
99
100	# get batch
101	batch = _parseJSON(self.request.get('batch', ''))	1✔
102	if batch and ('size' not in batch or 'page' not in batch):	1✔
103	batch = None # batching not providing correct options	×
104	if batch:	1✔
105	# must be slicable for batching support
106	page = int(batch['page'])	1✔
107	# page is being passed in is 1-based
108	start = (max(page - 1, 0)) * int(batch['size'])	1✔
109	end = start + int(batch['size'])	1✔
110	# Try __getitem__-based slice, then iterator slice.
111	# The iterator slice has to consume the iterator through
112	# to the desired slice, but that shouldn't be the end
113	# of the world because at some point the user will hopefully
114	# give up scrolling and search instead.
115	try:	1✔
116	results = results[start:end]	1✔
117	except TypeError:	1✔
118	results = itertools.islice(results, start, end)	1✔
119
120	# build result items
121	items = []	1✔
122
123	attributes = _parseJSON(self.request.get('attributes', ''))	1✔
124	if isinstance(attributes, basestring) and attributes:	1✔
125	attributes = attributes.split(',')	1✔
126
127	if attributes:	1✔
128	portal = getToolByName(context, 'portal_url').getPortalObject()	1✔
129	base_path = '/'.join(portal.getPhysicalPath())	1✔
130	for vocab_item in results:	1✔
131	if not results_are_brains:	1✔
132	vocab_item = vocab_item.value	1✔
133	item = {}	1✔
134	for attr in attributes:	1✔
135	key = attr	1✔
136	if ':' in attr:	1✔
137	key, attr = attr.split(':', 1)	×
138	if attr in _unsafe_metadata:	1✔
139	continue	×
140	if key == 'path':	1✔
141	attr = 'getPath'	1✔
142	val = getattr(vocab_item, attr, None)	1✔
143	if callable(val):	1✔
144	if attr in _safe_callable_metadata:	1✔
145	val = val()	1✔
146	else:
147	continue	×
148	if key == 'path':	1✔
149	val = val[len(base_path):]	1✔
150	item[key] = val	1✔
151	items.append(item)	1✔
152	else:
153	for item in results:	1✔
154	items.append({'id': item.token, 'text': item.title})	1✔
155
156	if total == 0:	1✔
157	total = len(items)	1✔
158
159	return json.dumps({	1✔
160	'results': items,
161	'total': total
162	})
163
164	def parsed_query(self, ):	1✔
165	query = _parseJSON(self.request.get('query', '')) or {}	1✔
166	if query:	1✔
167	parsed = queryparser.parseFormquery(	1✔
168	self.get_context(), query['criteria'])
169	if 'sort_on' in query:	1✔
170	parsed['sort_on'] = query['sort_on']	1✔
171	if 'sort_order' in query:	1✔
172	parsed['sort_order'] = str(query['sort_order'])	1✔
173	query = parsed	1✔
174	return query	1✔
175
176
177	class VocabularyView(BaseVocabularyView):	1✔
178	"""Queries a named vocabulary and returns JSON-formatted results."""
179
180	def get_context(self):	1✔
181	return self.context	1✔
182
183	def get_vocabulary(self):	1✔
184	# Look up named vocabulary and check permission.
185
186	context = self.context	1✔
187	factory_name = self.request.get('name', None)	1✔
188	field_name = self.request.get('field', None)	1✔
189	if not factory_name:	1✔
190	raise VocabLookupException('No factory provided.')	×
191	authorized = None	1✔
192	sm = getSecurityManager()	1✔
193	if factory_name in _permissions:	1✔
194	# Short circuit if permission is in global registry
195	authorized = sm.checkPermission(	1✔
196	_permissions[factory_name], context
197	)
198	elif field_name:	1✔
199	# Check field specific permission
200	permission_checker = queryAdapter(	1✔
201	context, IFieldPermissionChecker
202	)
203	if permission_checker is not None:	1✔
204	authorized = permission_checker.validate(	1✔
205	field_name, factory_name
206	)
207
208	if not authorized:	1✔
209	raise VocabLookupException('Vocabulary lookup not allowed')	1✔
210
211	factory = queryUtility(IVocabularyFactory, factory_name)	1✔
212	if not factory:	1✔
213	raise VocabLookupException(	1✔
214	'No factory with name "%s" exists.' % factory_name)
215
216	# This part is for backwards-compatibility with the first
217	# generation of vocabularies created for plone.app.widgets,
218	# which take the (unparsed) query as a parameter of the vocab
219	# factory rather than as a separate search method.
220	if type(factory) is FunctionType:	1✔
221	factory_spec = inspect.getargspec(factory)	1✔
222	else:
223	factory_spec = inspect.getargspec(factory.__call__)	1✔
224	query = _parseJSON(self.request.get('query', ''))	1✔
225	if query and 'query' in factory_spec.args:	1✔
226	vocabulary = factory(context, query=query)	1✔
227	else:
228	# This is what is reached for non-legacy vocabularies.
229	vocabulary = factory(context)	1✔
230
231	return vocabulary	1✔
232
233
234	class SourceView(BaseVocabularyView):	1✔
235	"""Queries a field's source and returns JSON-formatted results."""
236
237	def get_context(self):	1✔
238	return self.context.context	1✔
239
240	def get_vocabulary(self):	1✔
241	widget = self.context	1✔
242	field = widget.field.bind(widget.context)	1✔
243
244	# check field's write permission
245	info = mergedTaggedValueDict(field.interface, WRITE_PERMISSIONS_KEY)	1✔
246	permission_name = info.get(field.__name__, 'cmf.ModifyPortalContent')	1✔
247	permission = queryUtility(IPermission, name=permission_name)	1✔
248	if permission is None:	1✔
249	permission = getUtility(	×
250	IPermission, name='cmf.ModifyPortalContent')
251	if not getSecurityManager().checkPermission(	1✔
252	permission.title, self.get_context()):
253	raise VocabLookupException('Vocabulary lookup not allowed.')	1✔
254
255	if ICollection.providedBy(field):	1✔
256	return field.value_type.vocabulary	1✔
257	else:
258	return field.vocabulary	1✔

plone / plone.app.widgets / 665

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous