Merge pull request #2393 from Pylons/feature.invoke_exception_view

request.invoke exception view

docs/api/request.rst

@@ -13,7 +13,8 @@
                      current_route_path, static_url, static_path,
                      model_url, resource_url, resource_path, set_property, 
                      effective_principals, authenticated_userid,
-                     unauthenticated_userid, has_permission
+                     unauthenticated_userid, has_permission,
+                     invoke_exception_view
 
    .. attribute:: context
 
@@ -259,6 +260,8 @@
 
           See also :ref:`subrequest_chapter`.
 
+   .. automethod:: invoke_exception_view
+
    .. automethod:: has_permission
 
    .. automethod:: add_response_callback

pyramid/renderers.py

@@ -1,4 +1,3 @@
-import contextlib
 import json
 import os
 import re
@@ -30,6 +29,7 @@
 
 from pyramid.response import _get_response_factory
 from pyramid.threadlocal import get_current_registry
+from pyramid.util import hide_attrs
 
 # API
 
@@ -77,7 +77,7 @@ def render(renderer_name, value, request=None, package=None):
     helper = RendererHelper(name=renderer_name, package=package,
                             registry=registry)
 
-    with temporary_response(request):
+    with hide_attrs(request, 'response'):
         result = helper.render(value, None, request=request)
 
     return result
@@ -138,30 +138,13 @@ def render_to_response(renderer_name,
     helper = RendererHelper(name=renderer_name, package=package,
                             registry=registry)
 
-    with temporary_response(request):
+    with hide_attrs(request, 'response'):
         if response is not None:
             request.response = response
         result = helper.render_to_response(value, None, request=request)
 
     return result
 
-_marker = object()
-
-@contextlib.contextmanager
-def temporary_response(request):
-    """
-    Temporarily delete request.response and restore it afterward.
-    """
-    attrs = request.__dict__ if request is not None else {}
-    saved_response = attrs.pop('response', _marker)
-    try:
-        yield
-    finally:
-        if saved_response is not _marker:
-            attrs['response'] = saved_response
-        elif 'response' in attrs:
-            del attrs['response']
-
 def get_renderer(renderer_name, package=None):
     """ Return the renderer object for the renderer ``renderer_name``.
 

pyramid/request.py

@@ -32,6 +32,7 @@
     InstancePropertyHelper,
     InstancePropertyMixin,
 )
+from pyramid.view import ViewMethodsMixin
 
 class TemplateContext(object):
     pass
@@ -154,6 +155,7 @@ class Request(
     LocalizerRequestMixin,
     AuthenticationAPIMixin,
     AuthorizationAPIMixin,
+    ViewMethodsMixin,
     ):
     """
     A subclass of the :term:`WebOb` Request class.  An instance of

pyramid/testing.py

@@ -41,6 +41,7 @@
 from pyramid.request import CallbackMethodsMixin
 from pyramid.url import URLMethodsMixin
 from pyramid.util import InstancePropertyMixin
+from pyramid.view import ViewMethodsMixin
 
 
 _marker = object()
@@ -293,6 +294,7 @@ class DummyRequest(
     LocalizerRequestMixin,
     AuthenticationAPIMixin,
     AuthorizationAPIMixin,
+    ViewMethodsMixin,
     ):
     """ A DummyRequest object (incompletely) imitates a :term:`request` object.
 

pyramid/tests/test_renderers.py

@@ -592,48 +592,6 @@ class DummyRequestWithClassResponse(object):
         self.assertEqual(result.body, b'{"a": 1}')
         self.assertFalse('response' in request.__dict__)
 
-class Test_temporary_response(unittest.TestCase):
-    def _callFUT(self, request):
-        from pyramid.renderers import temporary_response
-        return temporary_response(request)
-
-    def test_restores_response(self):
-        request = testing.DummyRequest()
-        orig_response = request.response
-        with self._callFUT(request):
-            request.response = object()
-        self.assertEqual(request.response, orig_response)
-
-    def test_restores_response_on_exception(self):
-        request = testing.DummyRequest()
-        orig_response = request.response
-        try:
-            with self._callFUT(request):
-                request.response = object()
-                raise RuntimeError()
-        except RuntimeError:
-            self.assertEqual(request.response, orig_response)
-        else:                   # pragma: no cover
-            self.fail("RuntimeError not raised")
-
-    def test_restores_response_to_none(self):
-        request = testing.DummyRequest(response=None)
-        with self._callFUT(request):
-            request.response = object()
-        self.assertEqual(request.response, None)
-
-    def test_deletes_response(self):
-        request = testing.DummyRequest()
-        with self._callFUT(request):
-            request.response = object()
-        self.assertTrue('response' not in request.__dict__)
-
-    def test_does_not_delete_response_if_no_response_to_delete(self):
-        request = testing.DummyRequest()
-        with self._callFUT(request):
-            pass
-        self.assertTrue('response' not in request.__dict__)
-
 class Test_get_renderer(unittest.TestCase):
     def setUp(self):
         self.config = testing.setUp()

pyramid/tests/test_util.py

@@ -794,6 +794,63 @@ def get_bad_name():
         self.assertRaises(ConfigurationError, get_bad_name)
 
 
+class Test_hide_attrs(unittest.TestCase):
+    def _callFUT(self, obj, *attrs):
+        from pyramid.util import hide_attrs
+        return hide_attrs(obj, *attrs)
+
+    def _makeDummy(self):
+        from pyramid.decorator import reify
+        class Dummy(object):
+            x = 1
+
+            @reify
+            def foo(self):
+                return self.x
+        return Dummy()
+
+    def test_restores_attrs(self):
+        obj = self._makeDummy()
+        obj.bar = 'asdf'
+        orig_foo = obj.foo
+        with self._callFUT(obj, 'foo', 'bar'):
+            obj.foo = object()
+            obj.bar = 'nope'
+        self.assertEqual(obj.foo, orig_foo)
+        self.assertEqual(obj.bar, 'asdf')
+
+    def test_restores_attrs_on_exception(self):
+        obj = self._makeDummy()
+        orig_foo = obj.foo
+        try:
+            with self._callFUT(obj, 'foo'):
+                obj.foo = object()
+                raise RuntimeError()
+        except RuntimeError:
+            self.assertEqual(obj.foo, orig_foo)
+        else:                   # pragma: no cover
+            self.fail("RuntimeError not raised")
+
+    def test_restores_attrs_to_none(self):
+        obj = self._makeDummy()
+        obj.foo = None
+        with self._callFUT(obj, 'foo'):
+            obj.foo = object()
+        self.assertEqual(obj.foo, None)
+
+    def test_deletes_attrs(self):
+        obj = self._makeDummy()
+        with self._callFUT(obj, 'foo'):
+            obj.foo = object()
+        self.assertTrue('foo' not in obj.__dict__)
+
+    def test_does_not_delete_attr_if_no_attr_to_delete(self):
+        obj = self._makeDummy()
+        with self._callFUT(obj, 'foo'):
+            pass
+        self.assertTrue('foo' not in obj.__dict__)
+
+
 def dummyfunc(): pass
 
 

pyramid/tests/test_view.py

@@ -673,6 +673,138 @@ class Foo(object): pass
         class Bar(Foo): pass
         self.assertEqual(Bar.__view_defaults__, {})
 
+class TestViewMethodsMixin(unittest.TestCase):
+    def setUp(self):
+        self.config = testing.setUp()
+
+    def tearDown(self):
+        testing.tearDown()
+
+    def _makeOne(self, environ=None):
+        from pyramid.decorator import reify
+        from pyramid.view import ViewMethodsMixin
+        if environ is None:
+            environ = {}
+        class Request(ViewMethodsMixin):
+            def __init__(self, environ):
+                self.environ = environ
+
+            @reify
+            def response(self):
+                return DummyResponse()
+        request = Request(environ)
+        request.registry = self.config.registry
+        return request
+
+    def test_it(self):
+        def exc_view(exc, request):
+            self.assertTrue(exc is dummy_exc)
+            self.assertTrue(request.exception is dummy_exc)
+            return DummyResponse(b'foo')
+        self.config.add_view(exc_view, context=RuntimeError)
+        request = self._makeOne()
+        dummy_exc = RuntimeError()
+        try:
+            raise dummy_exc
+        except RuntimeError:
+            response = request.invoke_exception_view()
+            self.assertEqual(response.app_iter, [b'foo'])
+        else: # pragma: no cover
+            self.fail()
+
+    def test_it_hides_attrs(self):
+        def exc_view(exc, request):
+            self.assertTrue(exc is not orig_exc)
+            self.assertTrue(request.exception is not orig_exc)
+            self.assertTrue(request.exc_info is not orig_exc_info)
+            self.assertTrue(request.response is not orig_response)
+            request.response.app_iter = [b'bar']
+            return request.response
+        self.config.add_view(exc_view, context=RuntimeError)
+        request = self._makeOne()
+        orig_exc = request.exception = DummyContext()
+        orig_exc_info = request.exc_info = DummyContext()
+        orig_response = request.response = DummyResponse(b'foo')
+        try:
+            raise RuntimeError
+        except RuntimeError:
+            response = request.invoke_exception_view()
+            self.assertEqual(response.app_iter, [b'bar'])
+            self.assertTrue(request.exception is orig_exc)
+            self.assertTrue(request.exc_info is orig_exc_info)
+            self.assertTrue(request.response is orig_response)
+        else: # pragma: no cover
+            self.fail()
+
+    def test_it_supports_alternate_requests(self):
+        def exc_view(exc, request):
+            self.assertTrue(request is other_req)
+            return DummyResponse(b'foo')
+        self.config.add_view(exc_view, context=RuntimeError)
+        request = self._makeOne()
+        other_req = self._makeOne()
+        try:
+            raise RuntimeError
+        except RuntimeError:
+            response = request.invoke_exception_view(request=other_req)
+            self.assertEqual(response.app_iter, [b'foo'])
+        else: # pragma: no cover
+            self.fail()
+
+    def test_it_supports_threadlocal_registry(self):
+        def exc_view(exc, request):
+            return DummyResponse(b'foo')
+        self.config.add_view(exc_view, context=RuntimeError)
+        request = self._makeOne()
+        del request.registry
+        try:
+            raise RuntimeError
+        except RuntimeError:
+            response = request.invoke_exception_view()
+            self.assertEqual(response.app_iter, [b'foo'])
+        else: # pragma: no cover
+            self.fail()
+
+    def test_it_supports_alternate_exc_info(self):
+        def exc_view(exc, request):
+            self.assertTrue(request.exc_info is exc_info)
+            return DummyResponse(b'foo')
+        self.config.add_view(exc_view, context=RuntimeError)
+        request = self._makeOne()
+        try:
+            raise RuntimeError
+        except RuntimeError:
+            exc_info = sys.exc_info()
+        response = request.invoke_exception_view(exc_info=exc_info)
+        self.assertEqual(response.app_iter, [b'foo'])
+
+    def test_it_rejects_secured_view(self):
+        from pyramid.exceptions import Forbidden
+        def exc_view(exc, request): pass
+        self.config.testing_securitypolicy(permissive=False)
+        self.config.add_view(exc_view, context=RuntimeError, permission='view')
+        request = self._makeOne()
+        try:
+            raise RuntimeError
+        except RuntimeError:
+            self.assertRaises(Forbidden, request.invoke_exception_view)
+        else: # pragma: no cover
+            self.fail()
+
+    def test_it_allows_secured_view(self):
+        def exc_view(exc, request):
+            return DummyResponse(b'foo')
+        self.config.testing_securitypolicy(permissive=False)
+        self.config.add_view(exc_view, context=RuntimeError, permission='view')
+        request = self._makeOne()
+        try:
+            raise RuntimeError
+        except RuntimeError:
+            response = request.invoke_exception_view(secure=False)
+            self.assertEqual(response.app_iter, [b'foo'])
+        else: # pragma: no cover
+            self.fail()
+
 class ExceptionResponse(Exception):
     status = '404 Not Found'
     app_iter = ['Not Found']

pyramid/util.py

@@ -1,3 +1,4 @@
+import contextlib
 import functools
 try:
     # py2.7.7+ and py3.3+ have native comparison support
@@ -591,3 +592,22 @@ def get_callable_name(name):
             'used on __name__ of the method'
         )
         raise ConfigurationError(msg % name)
+
+@contextlib.contextmanager
+def hide_attrs(obj, *attrs):
+    """
+    Temporarily delete object attrs and restore afterward.
+    """
+    obj_vals = obj.__dict__ if obj is not None else {}
+    saved_vals = {}
+    for name in attrs:
+        saved_vals[name] = obj_vals.pop(name, _marker)
+    try:
+        yield
+    finally:
+        for name in attrs:
+            saved_val = saved_vals[name]
+            if saved_val is not _marker:
+                obj_vals[name] = saved_val
+            elif name in obj_vals:
+                del obj_vals[name]