tox.ini: Measure test coveragetox_coverage

1 files changed, 389 insertions, 0 deletions

diff --git a/paste/errordocument.py b/paste/errordocument.py
new file mode 100644
index 0000000..34f2d4a
--- /dev/null
+++ b/paste/errordocument.py
@@ -0,0 +1,389 @@
+# (c) 2005-2006 James Gardner <james@pythonweb.org>
+# This module is part of the Python Paste Project and is released under
+# the MIT License: http://www.opensource.org/licenses/mit-license.php
+"""
+Middleware to display error documents for certain status codes
+
+The middleware in this module can be used to intercept responses with
+specified status codes and internally forward the request to an appropriate
+URL where the content can be displayed to the user as an error document.
+"""
+
+import warnings
+import sys
+from six.moves.urllib import parse as urlparse
+from paste.recursive import ForwardRequestException, RecursiveMiddleware, RecursionLoop
+from paste.util import converters
+from paste.response import replace_header
+import six
+
+def forward(app, codes):
+    """
+    Intercepts a response with a particular status code and returns the
+    content from a specified URL instead.
+
+    The arguments are:
+
+    ``app``
+        The WSGI application or middleware chain.
+
+    ``codes``
+        A dictionary of integer status codes and the URL to be displayed
+        if the response uses that code.
+
+    For example, you might want to create a static file to display a
+    "File Not Found" message at the URL ``/error404.html`` and then use
+    ``forward`` middleware to catch all 404 status codes and display the page
+    you created. In this example ``app`` is your exisiting WSGI
+    applicaiton::
+
+        from paste.errordocument import forward
+        app = forward(app, codes={404:'/error404.html'})
+
+    """
+    for code in codes:
+        if not isinstance(code, int):
+            raise TypeError('All status codes should be type int. '
+                '%s is not valid'%repr(code))
+
+    def error_codes_mapper(code, message, environ, global_conf, codes):
+        if code in codes:
+            return codes[code]
+        else:
+            return None
+
+    #return _StatusBasedRedirect(app, error_codes_mapper, codes=codes)
+    return RecursiveMiddleware(
+        StatusBasedForward(
+            app,
+            error_codes_mapper,
+            codes=codes,
+        )
+    )
+
+class StatusKeeper(object):
+    def __init__(self, app, status, url, headers):
+        self.app = app
+        self.status = status
+        self.url = url
+        self.headers = headers
+
+    def __call__(self, environ, start_response):
+        def keep_status_start_response(status, headers, exc_info=None):
+            for header, value in headers:
+                if header.lower() == 'set-cookie':
+                    self.headers.append((header, value))
+                else:
+                    replace_header(self.headers, header, value)
+            return start_response(self.status, self.headers, exc_info)
+        parts = self.url.split('?')
+        environ['PATH_INFO'] = parts[0]
+        if len(parts) > 1:
+            environ['QUERY_STRING'] = parts[1]
+        else:
+            environ['QUERY_STRING'] = ''
+        #raise Exception(self.url, self.status)
+        try:
+            return self.app(environ, keep_status_start_response)
+        except RecursionLoop as e:
+            line = 'Recursion error getting error page: %s\n' % e
+            if six.PY3:
+                line = line.encode('utf8')
+            environ['wsgi.errors'].write(line)
+            keep_status_start_response('500 Server Error', [('Content-type', 'text/plain')], sys.exc_info())
+            body = ('Error: %s.  (Error page could not be fetched)'
+                    % self.status)
+            if six.PY3:
+                body = body.encode('utf8')
+            return [body]
+
+
+class StatusBasedForward(object):
+    """
+    Middleware that lets you test a response against a custom mapper object to
+    programatically determine whether to internally forward to another URL and
+    if so, which URL to forward to.
+
+    If you don't need the full power of this middleware you might choose to use
+    the simpler ``forward`` middleware instead.
+
+    The arguments are:
+
+    ``app``
+        The WSGI application or middleware chain.
+
+    ``mapper``
+        A callable that takes a status code as the
+        first parameter, a message as the second, and accepts optional environ,
+        global_conf and named argments afterwards. It should return a
+        URL to forward to or ``None`` if the code is not to be intercepted.
+
+    ``global_conf``
+        Optional default configuration from your config file. If ``debug`` is
+        set to ``true`` a message will be written to ``wsgi.errors`` on each
+        internal forward stating the URL forwarded to.
+
+    ``**params``
+        Optional, any other configuration and extra arguments you wish to
+        pass which will in turn be passed back to the custom mapper object.
+
+    Here is an example where a ``404 File Not Found`` status response would be
+    redirected to the URL ``/error?code=404&message=File%20Not%20Found``. This
+    could be useful for passing the status code and message into another
+    application to display an error document:
+
+    .. code-block:: python
+
+        from paste.errordocument import StatusBasedForward
+        from paste.recursive import RecursiveMiddleware
+        from urllib import urlencode
+
+        def error_mapper(code, message, environ, global_conf, kw)
+            if code in [404, 500]:
+                params = urlencode({'message':message, 'code':code})
+                url = '/error?'%(params)
+                return url
+            else:
+                return None
+
+        app = RecursiveMiddleware(
+            StatusBasedForward(app, mapper=error_mapper),
+        )
+
+    """
+
+    def __init__(self, app, mapper, global_conf=None, **params):
+        if global_conf is None:
+            global_conf = {}
+        # @@: global_conf shouldn't really come in here, only in a
+        # separate make_status_based_forward function
+        if global_conf:
+            self.debug = converters.asbool(global_conf.get('debug', False))
+        else:
+            self.debug = False
+        self.application = app
+        self.mapper = mapper
+        self.global_conf = global_conf
+        self.params = params
+
+    def __call__(self, environ, start_response):
+        url = []
+
+        def change_response(status, headers, exc_info=None):
+            status_code = status.split(' ')
+            try:
+                code = int(status_code[0])
+            except (ValueError, TypeError):
+                raise Exception(
+                    'StatusBasedForward middleware '
+                    'received an invalid status code %s'%repr(status_code[0])
+                )
+            message = ' '.join(status_code[1:])
+            new_url = self.mapper(
+                code,
+                message,
+                environ,
+                self.global_conf,
+                **self.params
+            )
+            if not (new_url == None or isinstance(new_url, str)):
+                raise TypeError(
+                    'Expected the url to internally '
+                    'redirect to in the StatusBasedForward mapper'
+                    'to be a string or None, not %r' % new_url)
+            if new_url:
+                url.append([new_url, status, headers])
+                # We have to allow the app to write stuff, even though
+                # we'll ignore it:
+                return [].append
+            else:
+                return start_response(status, headers, exc_info)
+
+        app_iter = self.application(environ, change_response)
+        if url:
+            if hasattr(app_iter, 'close'):
+                app_iter.close()
+
+            def factory(app):
+                return StatusKeeper(app, status=url[0][1], url=url[0][0],
+                                    headers=url[0][2])
+            raise ForwardRequestException(factory=factory)
+        else:
+            return app_iter
+
+def make_errordocument(app, global_conf, **kw):
+    """
+    Paste Deploy entry point to create a error document wrapper.
+
+    Use like::
+
+        [filter-app:main]
+        use = egg:Paste#errordocument
+        next = real-app
+        500 = /lib/msg/500.html
+        404 = /lib/msg/404.html
+    """
+    map = {}
+    for status, redir_loc in kw.items():
+        try:
+            status = int(status)
+        except ValueError:
+            raise ValueError('Bad status code: %r' % status)
+        map[status] = redir_loc
+    forwarder = forward(app, map)
+    return forwarder
+
+__pudge_all__ = [
+    'forward',
+    'make_errordocument',
+    'empty_error',
+    'make_empty_error',
+    'StatusBasedForward',
+]
+
+
+###############################################################################
+## Deprecated
+###############################################################################
+
+def custom_forward(app, mapper, global_conf=None, **kw):
+    """
+    Deprectated; use StatusBasedForward instead.
+    """
+    warnings.warn(
+        "errordocuments.custom_forward has been deprecated; please "
+        "use errordocuments.StatusBasedForward",
+        DeprecationWarning, 2)
+    if global_conf is None:
+        global_conf = {}
+    return _StatusBasedRedirect(app, mapper, global_conf, **kw)
+
+class _StatusBasedRedirect(object):
+    """
+    Deprectated; use StatusBasedForward instead.
+    """
+    def __init__(self, app, mapper, global_conf=None, **kw):
+
+        warnings.warn(
+            "errordocuments._StatusBasedRedirect has been deprecated; please "
+            "use errordocuments.StatusBasedForward",
+            DeprecationWarning, 2)
+
+        if global_conf is None:
+            global_conf = {}
+        self.application = app
+        self.mapper = mapper
+        self.global_conf = global_conf
+        self.kw = kw
+        self.fallback_template = """
+            <html>
+            <head>
+            <title>Error %(code)s</title>
+            </html>
+            <body>
+            <h1>Error %(code)s</h1>
+            <p>%(message)s</p>
+            <hr>
+            <p>
+                Additionally an error occurred trying to produce an
+                error document.  A description of the error was logged
+                to <tt>wsgi.errors</tt>.
+            </p>
+            </body>
+            </html>
+        """
+
+    def __call__(self, environ, start_response):
+        url = []
+        code_message = []
+        try:
+            def change_response(status, headers, exc_info=None):
+                new_url = None
+                parts = status.split(' ')
+                try:
+                    code = int(parts[0])
+                except (ValueError, TypeError):
+                    raise Exception(
+                        '_StatusBasedRedirect middleware '
+                        'received an invalid status code %s'%repr(parts[0])
+                    )
+                message = ' '.join(parts[1:])
+                new_url = self.mapper(
+                    code,
+                    message,
+                    environ,
+                    self.global_conf,
+                    self.kw
+                )
+                if not (new_url == None or isinstance(new_url, str)):
+                    raise TypeError(
+                        'Expected the url to internally '
+                        'redirect to in the _StatusBasedRedirect error_mapper'
+                        'to be a string or None, not %s'%repr(new_url)
+                    )
+                if new_url:
+                    url.append(new_url)
+                code_message.append([code, message])
+                return start_response(status, headers, exc_info)
+            app_iter = self.application(environ, change_response)
+        except:
+            try:
+                import sys
+                error = str(sys.exc_info()[1])
+            except:
+                error = ''
+            try:
+                code, message = code_message[0]
+            except:
+                code, message = ['', '']
+            environ['wsgi.errors'].write(
+                'Error occurred in _StatusBasedRedirect '
+                'intercepting the response: '+str(error)
+            )
+            return [self.fallback_template
+                    % {'message': message, 'code': code}]
+        else:
+            if url:
+                url_ = url[0]
+                new_environ = {}
+                for k, v in environ.items():
+                    if k != 'QUERY_STRING':
+                        new_environ['QUERY_STRING'] = urlparse.urlparse(url_)[4]
+                    else:
+                        new_environ[k] = v
+                class InvalidForward(Exception):
+                    pass
+                def eat_start_response(status, headers, exc_info=None):
+                    """
+                    We don't want start_response to do anything since it
+                    has already been called
+                    """
+                    if status[:3] != '200':
+                        raise InvalidForward(
+                            "The URL %s to internally forward "
+                            "to in order to create an error document did not "
+                            "return a '200' status code." % url_
+                        )
+                forward = environ['paste.recursive.forward']
+                old_start_response = forward.start_response
+                forward.start_response = eat_start_response
+                try:
+                    app_iter = forward(url_, new_environ)
+                except InvalidForward:
+                    code, message = code_message[0]
+                    environ['wsgi.errors'].write(
+                        'Error occurred in '
+                        '_StatusBasedRedirect redirecting '
+                        'to new URL: '+str(url[0])
+                    )
+                    return [
+                        self.fallback_template%{
+                            'message':message,
+                            'code':code,
+                        }
+                    ]
+                else:
+                    forward.start_response = old_start_response
+                    return app_iter
+            else:
+                return app_iter