3956162881

Build Type

push

github

Committed by Michael Howitz

Commit Message

Update to deprecation warning free releases.

Run Details

4401 of 7036 branches covered (62.55%)

Branch coverage included in aggregate %.

27161 of 31488 relevant lines covered (86.26%)

0.86 hits per line

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

68.0

/src/OFS/DTMLMethod.py

##############################################################################
#
# Copyright (c) 2002 Zope Foundation and Contributors.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE
#
##############################################################################
"""DTML Method objects.
"""
import re
from urllib.parse import quote

from AccessControl import getSecurityManager
from AccessControl.class_init import InitializeClass
from AccessControl.Permissions import change_proxy_roles  # NOQA
from AccessControl.Permissions import view as View
from AccessControl.Permissions import view_management_screens
from AccessControl.requestmethod import requestmethod
from AccessControl.SecurityInfo import ClassSecurityInfo
from AccessControl.tainted import TaintedString
from Acquisition import Implicit
from App.special_dtml import HTML
from App.special_dtml import DTMLFile
from DocumentTemplate.DT_Util import ParseError
from DocumentTemplate.permissions import change_dtml_methods
from DocumentTemplate.security import RestrictedDTML
from OFS.Cache import Cacheable
from OFS.History import Historical
from OFS.History import html_diff
from OFS.role import RoleManager
from OFS.SimpleItem import Item_w__name__
from OFS.SimpleItem import PathReprProvider
from zExceptions import Forbidden
from zExceptions import ResourceLockedError
from zExceptions.TracebackSupplement import PathTracebackSupplement
from zope.contenttype import guess_content_type
from ZPublisher.HTTPRequest import default_encoding
from ZPublisher.Iterators import IStreamIterator


_marker = []  # Create a new marker object.


class Code:
    # Documents masquerade as functions:
    pass


class DTMLMethod(
    PathReprProvider,
    RestrictedDTML,
    HTML,
    Implicit,
    RoleManager,
    Item_w__name__,
    Historical,
    Cacheable
):
    """ DocumentTemplate.HTML objects that act as methods of their containers.
    """
    meta_type = 'DTML Method'
    zmi_icon = 'far fa-file-alt'
    _proxy_roles = ()
    index_html = None  # Prevent accidental acquisition
    _cache_namespace_keys = ()
    _locked_error_text = 'This DTML Method is locked.'

    security = ClassSecurityInfo()
    security.declareObjectProtected(View)

    __code__ = Code()
    __code__.co_varnames = 'self', 'REQUEST', 'RESPONSE'
    __code__.co_argcount = 3
    __defaults__ = None

    manage_options = ((
        {
            'label': 'Edit',
            'action': 'manage_main',
        },
        {
            'label': 'View',
            'action': '',
        },
        {
            'label': 'Proxy',
            'action': 'manage_proxyForm',
        },
    ) + Historical.manage_options
      + RoleManager.manage_options
      + Item_w__name__.manage_options
      + Cacheable.manage_options
    )

    # Careful in permission changes--used by DTMLDocument!
    security.declareProtected(change_dtml_methods,  # NOQA: D001
                              'manage_historyCopy')
    security.declareProtected(change_dtml_methods,  # NOQA: D001
                              'manage_beforeHistoryCopy')
    security.declareProtected(change_dtml_methods,  # NOQA: D001
                              'manage_afterHistoryCopy')

    # More reasonable default for content-type for http HEAD requests.
    default_content_type = 'text/html'

    def errQuote(self, s):
        # Quoting is done when rendering the error in the template.
        return s

    @security.protected(View)
    def __call__(self, client=None, REQUEST={}, RESPONSE=None, **kw):
        """Render using the given client object

        o If client is not passed, we are being called as a sub-template:
          don't do any error propagation.

        o If supplied, use the REQUEST mapping, Response, and key word
        arguments.
        """
        if not self._cache_namespace_keys:
            data = self.ZCacheable_get(default=_marker)
            if data is not _marker:
                if IStreamIterator.isImplementedBy(data) and \
                   RESPONSE is not None:
                    # This is a stream iterator and we need to set some
                    # headers now before giving it to medusa
                    headers_get = RESPONSE.headers.get

                    if headers_get('content-length', None) is None:
                        RESPONSE.setHeader('content-length', len(data))

                    if headers_get('content-type', None) is None and \
                       headers_get('Content-type', None) is None:
                        ct = (self.__dict__.get('content_type')
                              or self.default_content_type)
                        RESPONSE.setHeader('content-type', ct)

                # Return cached results.
                return data

        __traceback_supplement__ = (PathTracebackSupplement, self)
        kw['document_id'] = self.getId()
        kw['document_title'] = self.title

        security = getSecurityManager()
        security.addContext(self)
        if 'validate' in self.__dict__:
            first_time_through = 0
        else:
            self.__dict__['validate'] = security.DTMLValidate
            first_time_through = 1
        try:

            if client is None:
                # Called as subtemplate, so don't need error propagation!
                r = HTML.__call__(self, client, REQUEST, **kw)
                if RESPONSE is None:
                    result = r
                else:
                    result = decapitate(r, RESPONSE)
                if not self._cache_namespace_keys:
                    self.ZCacheable_set(result)
                return result

            r = HTML.__call__(self, client, REQUEST, **kw)

            if RESPONSE is None or not isinstance(r, str):
                if not self._cache_namespace_keys:
                    self.ZCacheable_set(r)
                return r

        finally:
            security.removeContext(self)
            if first_time_through:
                del self.__dict__['validate']

        have_key = RESPONSE.headers.__contains__
        if not (have_key('content-type') or have_key('Content-Type')):
            if 'content_type' in self.__dict__:
                c = self.content_type
            else:
                encoding = getattr(self, 'encoding', default_encoding)
                c, e = guess_content_type(self.getId(), r.encode(encoding))
            RESPONSE.setHeader('Content-Type', c)
        result = decapitate(r, RESPONSE)
        if not self._cache_namespace_keys:
            self.ZCacheable_set(result)
        return result

    def validate(self, inst, parent, name, value, md=None):
        return getSecurityManager().validate(inst, parent, name, value)

    def ZDocumentTemplate_beforeRender(self, md, default):
        # Tries to get a cached value.
        if self._cache_namespace_keys:
            # Use the specified keys from the namespace to identify a
            # cache entry.
            kw = {}
            for key in self._cache_namespace_keys:
                try:
                    val = md[key]
                except Exception:
                    val = None
                kw[key] = val
            return self.ZCacheable_get(keywords=kw, default=default)
        return default

    def ZDocumentTemplate_afterRender(self, md, result):
        # Tries to set a cache value.
        if self._cache_namespace_keys:
            kw = {}
            for key in self._cache_namespace_keys:
                try:
                    val = md[key]
                except Exception:
                    val = None
                kw[key] = val
            self.ZCacheable_set(result, keywords=kw)

    security.declareProtected(change_dtml_methods, 'ZCacheable_configHTML')  # NOQA: D001,E501
    ZCacheable_configHTML = DTMLFile('dtml/cacheNamespaceKeys', globals())

    @security.protected(change_dtml_methods)
    def getCacheNamespaceKeys(self):
        # Return the cacheNamespaceKeys.
        return self._cache_namespace_keys

    @security.protected(change_dtml_methods)
    def setCacheNamespaceKeys(self, keys, REQUEST=None):
        # Set the list of names looked up to provide a cache key.
        ks = []
        for key in keys:
            key = str(key).strip()
            if key:
                ks.append(key)
        self._cache_namespace_keys = tuple(ks)

        if REQUEST is not None:
            return self.ZCacheable_manage(self, REQUEST)

    @security.protected(View)
    def get_size(self):
        return len(self.raw)

    # deprecated; use get_size!
    getSize = get_size

    security.declareProtected(change_dtml_methods, 'manage')  # NOQA: D001

    security.declareProtected(change_dtml_methods, 'manage_editForm')  # NOQA: D001,E501
    manage_editForm = DTMLFile('dtml/documentEdit', globals())
    manage_editForm._setName('manage_editForm')

    # deprecated!
    manage_uploadForm = manage_editForm

    security.declareProtected(change_dtml_methods, 'manage_main')  # NOQA: D001
    manage = manage_main = manage_editDocument = manage_editForm

    security.declareProtected(change_proxy_roles, 'manage_proxyForm')  # NOQA: D001,E501
    manage_proxyForm = DTMLFile('dtml/documentProxy', globals())

    @security.protected(change_dtml_methods)
    def manage_edit(self, data, title, SUBMIT='Change', REQUEST=None):
        """ Replace contents with 'data', title with 'title'.
        """
        self._validateProxy()
        if self.wl_isLocked():
            raise ResourceLockedError(self._locked_error_text)

        self.title = str(title)
        if isinstance(data, TaintedString):
            data = data.quoted()

        if hasattr(data, 'read'):
            data = data.read()
        try:
            self.munge(data)
        except ParseError as e:
            if REQUEST:
                return self.manage_main(
                    self, REQUEST, manage_tabs_message=e,
                    manage_tabs_type='warning')
            else:
                raise
        self.ZCacheable_invalidate()
        if REQUEST:
            message = "Saved changes."
            return self.manage_main(self, REQUEST, manage_tabs_message=message)

    @security.protected(change_dtml_methods)
    def manage_upload(self, file='', REQUEST=None):
        """ Replace the contents of the document with the text in 'file'.

        Store `file` as a native `str`.
        """
        self._validateProxy()
        if self.wl_isLocked():
            if REQUEST is not None:
                return self.manage_main(
                    self, REQUEST,
                    manage_tabs_message=self._locked_error_text,
                    manage_tabs_type='warning')
            raise ResourceLockedError(self._locked_error_text)

        if REQUEST is not None and not file:
            return self.manage_main(
                self, REQUEST,
                manage_tabs_message='No file specified',
                manage_tabs_type='warning')

        self.munge(safe_file_data(file))
        self.ZCacheable_invalidate()
        if REQUEST is not None:
            message = "Content uploaded."
            return self.manage_main(self, REQUEST, manage_tabs_message=message)

    def manage_haveProxy(self, r):
        return r in self._proxy_roles

    def _validateProxy(self, roles=None):
        if roles is None:
            roles = self._proxy_roles
        if not roles:
            return
        user = getSecurityManager().getUser()
        if user is not None and user.allowed(self, roles):
            return
        raise Forbidden(
            'You are not authorized to change <em>%s</em> because you '
            'do not have proxy roles.\n<!--%s, %s-->' % (
                self.__name__, user, roles))

    @security.protected(change_proxy_roles)
    @requestmethod('POST')
    def manage_proxy(self, roles=(), REQUEST=None):
        """Change Proxy Roles"""
        user = getSecurityManager().getUser()
        if 'Manager' not in user.getRolesInContext(self):
            self._validateProxy(roles)
            self._validateProxy()
        self.ZCacheable_invalidate()
        self._proxy_roles = tuple(roles)
        if REQUEST:
            message = "Saved changes."
            return self.manage_proxyForm(self, REQUEST,
                                         manage_tabs_message=message)

    @security.protected(view_management_screens)
    def PrincipiaSearchSource(self):
        # Support for searching - the document's contents are searched.
        return self.read()

    @security.protected(view_management_screens)
    def document_src(self, REQUEST=None, RESPONSE=None):
        # Return unprocessed document source.
        if RESPONSE is not None:
            RESPONSE.setHeader('Content-Type', 'text/plain')
        return self.read()

    @security.protected(change_dtml_methods)
    def PUT(self, REQUEST, RESPONSE):
        """ Handle HTTP PUT requests.
        """
        self.dav__init(REQUEST, RESPONSE)
        self.dav__simpleifhandler(REQUEST, RESPONSE, refresh=1)
        body = safe_file_data(REQUEST.get('BODY', ''))
        self._validateProxy()
        self.munge(body)
        self.ZCacheable_invalidate()
        RESPONSE.setStatus(204)
        return RESPONSE

    def manage_historyCompare(self, rev1, rev2, REQUEST,
                              historyComparisonResults=''):
        return DTMLMethod.inheritedAttribute('manage_historyCompare')(
            self, rev1, rev2, REQUEST,
            historyComparisonResults=html_diff(rev1.read(), rev2.read()))


InitializeClass(DTMLMethod)


token = r"[a-zA-Z0-9!#$%&'*+\-.\\\\^_`|~]+"
hdr_start = re.compile(r'(%s):(.*)' % token).match


def decapitate(html, RESPONSE=None):
    headers = []
    spos = 0
    eolen = 1
    while 1:
        m = hdr_start(html, spos)
        if not m:
            if html[spos:spos + 2] == '\r\n':
                eolen = 2
                break
            if html[spos:spos + 1] == '\n':
                eolen = 1
                break
            return html
        header = list(m.groups())
        headers.append(header)
        spos = m.end() + 1
        while spos < len(html) and html[spos] in ' \t':
            eol = html.find('\r\n', spos)
            if eol != -1:
                eolen = 2
            else:
                eol = html.find('\n', spos)
                if eol < 0:
                    return html
                eolen = 1
            header.append(html[spos:eol].strip())
            spos = eol + eolen
    if RESPONSE is not None:
        for header in headers:
            hkey = header.pop(0)
            RESPONSE.setHeader(hkey, ' '.join(header).strip())
    return html[spos + eolen:]


def safe_file_data(data):
    # Helper to convert upload file content into a safe value for saving
    if hasattr(data, 'read'):
        data = data.read()
    if isinstance(data, bytes):
        data = data.decode('utf-8')
    return data


default_dm_html = """\
<!DOCTYPE html>
<html>
  <head>
    <title><dtml-var title_or_id></title>
    <meta charset="utf-8" />
  </head>
  <body>
    <h2><dtml-var title_or_id> <dtml-var document_title></h2>
    <p>
    This is the <dtml-var document_id> Document
    in the <dtml-var title_and_id> Folder.
    </p>
  </body>
</html>"""

addForm = DTMLFile('dtml/methodAdd', globals())


def addDTMLMethod(self, id, title='', file='', REQUEST=None, submit=None):
    """Add a DTML Method object with the contents of file. If
    'file' is empty, default document text is used.
    """
    data = safe_file_data(file)
    if not data:
        data = default_dm_html
    id = str(id)
    title = str(title)
    ob = DTMLMethod(data, __name__=id)
    ob.title = title
    id = self._setObject(id, ob)
    if REQUEST is not None:
        try:
            u = self.DestinationURL()
        except Exception:
            u = REQUEST['URL1']
        if submit == "Add and Edit":
            u = f"{u}/{quote(id)}"
        REQUEST.RESPONSE.redirect(u + '/manage_main')
    return ''

1	##############################################################################
2	#
3	# Copyright (c) 2002 Zope Foundation and Contributors.
4	#
5	# This software is subject to the provisions of the Zope Public License,
6	# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
7	# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
8	# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
9	# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
10	# FOR A PARTICULAR PURPOSE
11	#
12	##############################################################################
13	"""DTML Method objects.	1✔
14	"""
15	import re	1✔
16	from urllib.parse import quote	1✔
17
18	from AccessControl import getSecurityManager	1✔
19	from AccessControl.class_init import InitializeClass	1✔
20	from AccessControl.Permissions import change_proxy_roles # NOQA	1✔
21	from AccessControl.Permissions import view as View	1✔
22	from AccessControl.Permissions import view_management_screens	1✔
23	from AccessControl.requestmethod import requestmethod	1✔
24	from AccessControl.SecurityInfo import ClassSecurityInfo	1✔
25	from AccessControl.tainted import TaintedString	1✔
26	from Acquisition import Implicit	1✔
27	from App.special_dtml import HTML	1✔
28	from App.special_dtml import DTMLFile	1✔
29	from DocumentTemplate.DT_Util import ParseError	1✔
30	from DocumentTemplate.permissions import change_dtml_methods	1✔
31	from DocumentTemplate.security import RestrictedDTML	1✔
32	from OFS.Cache import Cacheable	1✔
33	from OFS.History import Historical	1✔
34	from OFS.History import html_diff	1✔
35	from OFS.role import RoleManager	1✔
36	from OFS.SimpleItem import Item_w__name__	1✔
37	from OFS.SimpleItem import PathReprProvider	1✔
38	from zExceptions import Forbidden	1✔
39	from zExceptions import ResourceLockedError	1✔
40	from zExceptions.TracebackSupplement import PathTracebackSupplement	1✔
41	from zope.contenttype import guess_content_type	1✔
42	from ZPublisher.HTTPRequest import default_encoding	1✔
43	from ZPublisher.Iterators import IStreamIterator	1✔
44
45
46	_marker = [] # Create a new marker object.	1✔
47
48
49	class Code:	1✔
50	# Documents masquerade as functions:
51	pass	1✔
52
53
54	class DTMLMethod(	1✔
55	PathReprProvider,
56	RestrictedDTML,
57	HTML,
58	Implicit,
59	RoleManager,
60	Item_w__name__,
61	Historical,
62	Cacheable
63	):
64	""" DocumentTemplate.HTML objects that act as methods of their containers.
65	"""
66	meta_type = 'DTML Method'	1✔
67	zmi_icon = 'far fa-file-alt'	1✔
68	_proxy_roles = ()	1✔
69	index_html = None # Prevent accidental acquisition	1✔
70	_cache_namespace_keys = ()	1✔
71	_locked_error_text = 'This DTML Method is locked.'	1✔
72
73	security = ClassSecurityInfo()	1✔
74	security.declareObjectProtected(View)	1✔
75
76	__code__ = Code()	1✔
77	__code__.co_varnames = 'self', 'REQUEST', 'RESPONSE'	1✔
78	__code__.co_argcount = 3	1✔
79	__defaults__ = None	1✔
80
81	manage_options = ((	1✔
82	{
83	'label': 'Edit',
84	'action': 'manage_main',
85	},
86	{
87	'label': 'View',
88	'action': '',
89	},
90	{
91	'label': 'Proxy',
92	'action': 'manage_proxyForm',
93	},
94	) + Historical.manage_options
95	+ RoleManager.manage_options
96	+ Item_w__name__.manage_options
97	+ Cacheable.manage_options
98	)
99
100	# Careful in permission changes--used by DTMLDocument!
101	security.declareProtected(change_dtml_methods, # NOQA: D001	1✔
102	'manage_historyCopy')
103	security.declareProtected(change_dtml_methods, # NOQA: D001	1✔
104	'manage_beforeHistoryCopy')
105	security.declareProtected(change_dtml_methods, # NOQA: D001	1✔
106	'manage_afterHistoryCopy')
107
108	# More reasonable default for content-type for http HEAD requests.
109	default_content_type = 'text/html'	1✔
110
111	def errQuote(self, s):	1✔
112	# Quoting is done when rendering the error in the template.
113	return s	1✔
114
115	@security.protected(View)	1✔
116	def __call__(self, client=None, REQUEST={}, RESPONSE=None, **kw):	1✔
117	"""Render using the given client object
118
119	o If client is not passed, we are being called as a sub-template:
120	don't do any error propagation.
121
122	o If supplied, use the REQUEST mapping, Response, and key word
123	arguments.
124	"""
125	if not self._cache_namespace_keys:	1!
126	data = self.ZCacheable_get(default=_marker)	1✔
127	if data is not _marker:	1!
128	if IStreamIterator.isImplementedBy(data) and \	×
129	RESPONSE is not None:
130	# This is a stream iterator and we need to set some
131	# headers now before giving it to medusa
132	headers_get = RESPONSE.headers.get	×
133
134	if headers_get('content-length', None) is None:	×
135	RESPONSE.setHeader('content-length', len(data))	×
136
137	if headers_get('content-type', None) is None and \	×
138	headers_get('Content-type', None) is None:
139	ct = (self.__dict__.get('content_type')	×
140	or self.default_content_type)
141	RESPONSE.setHeader('content-type', ct)	×
142
143	# Return cached results.
144	return data	×
145
146	__traceback_supplement__ = (PathTracebackSupplement, self)	1✔
147	kw['document_id'] = self.getId()	1✔
148	kw['document_title'] = self.title	1✔
149
150	security = getSecurityManager()	1✔
151	security.addContext(self)	1✔
152	if 'validate' in self.__dict__:	1!
153	first_time_through = 0	×
154	else:
155	self.__dict__['validate'] = security.DTMLValidate	1✔
156	first_time_through = 1	1✔
157	try:	1✔
158
159	if client is None:	1!
160	# Called as subtemplate, so don't need error propagation!
161	r = HTML.__call__(self, client, REQUEST, **kw)	×
162	if RESPONSE is None:	×
163	result = r	×
164	else:
165	result = decapitate(r, RESPONSE)	×
166	if not self._cache_namespace_keys:	×
167	self.ZCacheable_set(result)	×
168	return result	×
169
170	r = HTML.__call__(self, client, REQUEST, **kw)	1✔
171
172	if RESPONSE is None or not isinstance(r, str):	1✔
173	if not self._cache_namespace_keys:	1!
174	self.ZCacheable_set(r)	1✔
175	return r	1✔
176
177	finally:
178	security.removeContext(self)	1✔
179	if first_time_through:	1!
180	del self.__dict__['validate']	1!
181
182	have_key = RESPONSE.headers.__contains__	1✔
183	if not (have_key('content-type') or have_key('Content-Type')):	1!
184	if 'content_type' in self.__dict__:	1!
185	c = self.content_type	×
186	else:
187	encoding = getattr(self, 'encoding', default_encoding)	1✔
188	c, e = guess_content_type(self.getId(), r.encode(encoding))	1✔
189	RESPONSE.setHeader('Content-Type', c)	1✔
190	result = decapitate(r, RESPONSE)	1✔
191	if not self._cache_namespace_keys:	1!
192	self.ZCacheable_set(result)	1✔
193	return result	1✔
194
195	def validate(self, inst, parent, name, value, md=None):	1✔
196	return getSecurityManager().validate(inst, parent, name, value)	×
197
198	def ZDocumentTemplate_beforeRender(self, md, default):	1✔
199	# Tries to get a cached value.
200	if self._cache_namespace_keys:	1!
201	# Use the specified keys from the namespace to identify a
202	# cache entry.
203	kw = {}	×
204	for key in self._cache_namespace_keys:	×
205	try:	×
206	val = md[key]	×
207	except Exception:	×
208	val = None	×
209	kw[key] = val	×
210	return self.ZCacheable_get(keywords=kw, default=default)	×
211	return default	1✔
212
213	def ZDocumentTemplate_afterRender(self, md, result):	1✔
214	# Tries to set a cache value.
215	if self._cache_namespace_keys:	1!
216	kw = {}	×
217	for key in self._cache_namespace_keys:	×
218	try:	×
219	val = md[key]	×
220	except Exception:	×
221	val = None	×
222	kw[key] = val	×
223	self.ZCacheable_set(result, keywords=kw)	×
224
225	security.declareProtected(change_dtml_methods, 'ZCacheable_configHTML') # NOQA: D001,E501	1✔
226	ZCacheable_configHTML = DTMLFile('dtml/cacheNamespaceKeys', globals())	1✔
227
228	@security.protected(change_dtml_methods)	1✔
229	def getCacheNamespaceKeys(self):	1✔
230	# Return the cacheNamespaceKeys.
231	return self._cache_namespace_keys	×
232
233	@security.protected(change_dtml_methods)	1✔
234	def setCacheNamespaceKeys(self, keys, REQUEST=None):	1✔
235	# Set the list of names looked up to provide a cache key.
236	ks = []	×
237	for key in keys:	×
238	key = str(key).strip()	×
239	if key:	×
240	ks.append(key)	×
241	self._cache_namespace_keys = tuple(ks)	×
242
243	if REQUEST is not None:	×
244	return self.ZCacheable_manage(self, REQUEST)	×
245
246	@security.protected(View)	1✔
247	def get_size(self):	1✔
248	return len(self.raw)	1✔
249
250	# deprecated; use get_size!
251	getSize = get_size	1✔
252
253	security.declareProtected(change_dtml_methods, 'manage') # NOQA: D001	1✔
254
255	security.declareProtected(change_dtml_methods, 'manage_editForm') # NOQA: D001,E501	1✔
256	manage_editForm = DTMLFile('dtml/documentEdit', globals())	1✔
257	manage_editForm._setName('manage_editForm')	1✔
258
259	# deprecated!
260	manage_uploadForm = manage_editForm	1✔
261
262	security.declareProtected(change_dtml_methods, 'manage_main') # NOQA: D001	1✔
263	manage = manage_main = manage_editDocument = manage_editForm	1✔
264
265	security.declareProtected(change_proxy_roles, 'manage_proxyForm') # NOQA: D001,E501	1✔
266	manage_proxyForm = DTMLFile('dtml/documentProxy', globals())	1✔
267
268	@security.protected(change_dtml_methods)	1✔
269	def manage_edit(self, data, title, SUBMIT='Change', REQUEST=None):	1✔
270	""" Replace contents with 'data', title with 'title'.
271	"""
272	self._validateProxy()	1✔
273	if self.wl_isLocked():	1✔
274	raise ResourceLockedError(self._locked_error_text)	1✔
275
276	self.title = str(title)	1✔
277	if isinstance(data, TaintedString):	1✔
278	data = data.quoted()	1✔
279
280	if hasattr(data, 'read'):	1!
281	data = data.read()	×
282	try:	1✔
283	self.munge(data)	1✔
284	except ParseError as e:	1✔
285	if REQUEST:	1✔
286	return self.manage_main(	1✔
287	self, REQUEST, manage_tabs_message=e,
288	manage_tabs_type='warning')
289	else:
290	raise	1✔
291	self.ZCacheable_invalidate()	1✔
292	if REQUEST:	1✔
293	message = "Saved changes."	1✔
294	return self.manage_main(self, REQUEST, manage_tabs_message=message)	1✔
295
296	@security.protected(change_dtml_methods)	1✔
297	def manage_upload(self, file='', REQUEST=None):	1✔
298	""" Replace the contents of the document with the text in 'file'.
299
300	Store `file` as a native `str`.
301	"""
302	self._validateProxy()	1✔
303	if self.wl_isLocked():	1✔
304	if REQUEST is not None:	1✔
305	return self.manage_main(	1✔
306	self, REQUEST,
307	manage_tabs_message=self._locked_error_text,
308	manage_tabs_type='warning')
309	raise ResourceLockedError(self._locked_error_text)	1✔
310
311	if REQUEST is not None and not file:	1✔
312	return self.manage_main(	1✔
313	self, REQUEST,
314	manage_tabs_message='No file specified',
315	manage_tabs_type='warning')
316
317	self.munge(safe_file_data(file))	1✔
318	self.ZCacheable_invalidate()	1✔
319	if REQUEST is not None:	1✔
320	message = "Content uploaded."	1✔
321	return self.manage_main(self, REQUEST, manage_tabs_message=message)	1✔
322
323	def manage_haveProxy(self, r):	1✔
324	return r in self._proxy_roles	1✔
325
326	def _validateProxy(self, roles=None):	1✔
327	if roles is None:	1✔
328	roles = self._proxy_roles	1✔
329	if not roles:	1✔
330	return	1✔
331	user = getSecurityManager().getUser()	1✔
332	if user is not None and user.allowed(self, roles):	1✔
333	return	1✔
334	raise Forbidden(	1✔
335	'You are not authorized to change <em>%s</em> because you '
336	'do not have proxy roles.\n<!--%s, %s-->' % (
337	self.__name__, user, roles))
338
339	@security.protected(change_proxy_roles)	1✔
340	@requestmethod('POST')	1✔
341	def manage_proxy(self, roles=(), REQUEST=None):	1✔
342	"""Change Proxy Roles"""
343	user = getSecurityManager().getUser()	1✔
344	if 'Manager' not in user.getRolesInContext(self):	1✔
345	self._validateProxy(roles)	1✔
346	self._validateProxy()	1✔
347	self.ZCacheable_invalidate()	1✔
348	self._proxy_roles = tuple(roles)	1✔
349	if REQUEST:	1!
350	message = "Saved changes."	1✔
351	return self.manage_proxyForm(self, REQUEST,	1✔
352	manage_tabs_message=message)
353
354	@security.protected(view_management_screens)	1✔
355	def PrincipiaSearchSource(self):	1✔
356	# Support for searching - the document's contents are searched.
357	return self.read()	×
358
359	@security.protected(view_management_screens)	1✔
360	def document_src(self, REQUEST=None, RESPONSE=None):	1✔
361	# Return unprocessed document source.
362	if RESPONSE is not None:	×
363	RESPONSE.setHeader('Content-Type', 'text/plain')	×
364	return self.read()	×
365
366	@security.protected(change_dtml_methods)	1✔
367	def PUT(self, REQUEST, RESPONSE):	1✔
368	""" Handle HTTP PUT requests.
369	"""
370	self.dav__init(REQUEST, RESPONSE)	1✔
371	self.dav__simpleifhandler(REQUEST, RESPONSE, refresh=1)	1✔
372	body = safe_file_data(REQUEST.get('BODY', ''))	1✔
373	self._validateProxy()	1✔
374	self.munge(body)	1✔
375	self.ZCacheable_invalidate()	1✔
376	RESPONSE.setStatus(204)	1✔
377	return RESPONSE	1✔
378
379	def manage_historyCompare(self, rev1, rev2, REQUEST,	1✔
380	historyComparisonResults=''):
381	return DTMLMethod.inheritedAttribute('manage_historyCompare')(	×
382	self, rev1, rev2, REQUEST,
383	historyComparisonResults=html_diff(rev1.read(), rev2.read()))
384
385
386	InitializeClass(DTMLMethod)	1✔
387
388
389	token = r"[a-zA-Z0-9!#$%&'*+\-.\\\\^_`\|~]+"	1✔
390	hdr_start = re.compile(r'(%s):(.*)' % token).match	1✔
391
392
393	def decapitate(html, RESPONSE=None):	1✔
394	headers = []	1✔
395	spos = 0	1✔
396	eolen = 1	1✔
397	while 1:
398	m = hdr_start(html, spos)	1✔
399	if not m:	1✔
400	if html[spos:spos + 2] == '\r\n':	1!
401	eolen = 2	×
402	break	×
403	if html[spos:spos + 1] == '\n':	1!
404	eolen = 1	×
405	break	×
406	return html	1✔
407	header = list(m.groups())	1✔
408	headers.append(header)	1✔
409	spos = m.end() + 1	1✔
410	while spos < len(html) and html[spos] in ' \t':	1!
411	eol = html.find('\r\n', spos)	×
412	if eol != -1:	×
413	eolen = 2	×
414	else:
415	eol = html.find('\n', spos)	×
416	if eol < 0:	×
417	return html	×
418	eolen = 1	×
419	header.append(html[spos:eol].strip())	×
420	spos = eol + eolen	×
421	if RESPONSE is not None:	×
422	for header in headers:	×
423	hkey = header.pop(0)	×
424	RESPONSE.setHeader(hkey, ' '.join(header).strip())	×
425	return html[spos + eolen:]	×
426
427
428	def safe_file_data(data):	1✔
429	# Helper to convert upload file content into a safe value for saving
430	if hasattr(data, 'read'):	1✔
431	data = data.read()	1✔
432	if isinstance(data, bytes):	1✔
433	data = data.decode('utf-8')	1✔
434	return data	1✔
435
436
437	default_dm_html = """\	1✔
438	<!DOCTYPE html>
439	<html>
440	<head>
441	<title><dtml-var title_or_id></title>
442	<meta charset="utf-8" />
443	</head>
444	<body>
445	<h2><dtml-var title_or_id> <dtml-var document_title></h2>
446	<p>
447	This is the <dtml-var document_id> Document
448	in the <dtml-var title_and_id> Folder.
449	</p>
450	</body>
451	</html>"""
452
453	addForm = DTMLFile('dtml/methodAdd', globals())	1✔
454
455
456	def addDTMLMethod(self, id, title='', file='', REQUEST=None, submit=None):	1✔
457	"""Add a DTML Method object with the contents of file. If
458	'file' is empty, default document text is used.
459	"""
460	data = safe_file_data(file)	1✔
461	if not data:	1✔
462	data = default_dm_html	1✔
463	id = str(id)	1✔
464	title = str(title)	1✔
465	ob = DTMLMethod(data, __name__=id)	1✔
466	ob.title = title	1✔
467	id = self._setObject(id, ob)	1✔
468	if REQUEST is not None:	1!
469	try:	×
470	u = self.DestinationURL()	×
471	except Exception:	×
472	u = REQUEST['URL1']	×
473	if submit == "Add and Edit":	×
474	u = f"{u}/{quote(id)}"	×
475	REQUEST.RESPONSE.redirect(u + '/manage_main')	×
476	return ''	1✔

zopefoundation / Zope / 3956162881

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