diff options
| -rw-r--r-- | CHANGES.txt | 9 | ||||
| -rw-r--r-- | docs/api/view.rst | 12 | ||||
| -rw-r--r-- | docs/index.rst | 1 | ||||
| -rw-r--r-- | repoze/bfg/interfaces.py | 1 | ||||
| -rw-r--r-- | repoze/bfg/router.py | 47 | ||||
| -rw-r--r-- | repoze/bfg/security.py | 16 | ||||
| -rw-r--r-- | repoze/bfg/tests/test_router.py | 16 | ||||
| -rw-r--r-- | repoze/bfg/tests/test_view.py | 284 | ||||
| -rw-r--r-- | repoze/bfg/view.py | 71 |
9 files changed, 408 insertions, 49 deletions
diff --git a/CHANGES.txt b/CHANGES.txt index 78bd4bb2f..5fe8c0fdf 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,7 +1,14 @@ +Next release + Features - Event notification is issued after application is created and - configured. + configured (``IWSGIApplicationCreatedEvent``). + + - New API module: ``repoze.bfg.view``. This module contains the functions + named ``render_view_to_response``, ``render_view_to_iterable`` and + ``is_response``, which are documented in the API docs. These features + aid programmatic (non-request-driven) view execution. 0.3.4 (08/28/2008) diff --git a/docs/api/view.rst b/docs/api/view.rst new file mode 100644 index 000000000..d7d238154 --- /dev/null +++ b/docs/api/view.rst @@ -0,0 +1,12 @@ +.. _view_module: + +:mod:`repoze.bfg.view` +---------------------- + +.. automodule:: repoze.bfg.view + + .. autofunction:: render_view_to_response + + .. autofunction:: render_view_to_iterable + + .. autofunction:: is_response diff --git a/docs/index.rst b/docs/index.rst index b4bbe4398..3d18c115f 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -61,6 +61,7 @@ Per-module :mod:`repoze.bfg` API documentation. api/template api/traversal api/urldispatch + api/view api/wsgi Sample Applications diff --git a/repoze/bfg/interfaces.py b/repoze/bfg/interfaces.py index db2d1622d..6c28c75a7 100644 --- a/repoze/bfg/interfaces.py +++ b/repoze/bfg/interfaces.py @@ -1,4 +1,3 @@ -from zope.interface import implements from zope.interface import Interface from zope.interface import Attribute diff --git a/repoze/bfg/router.py b/repoze/bfg/router.py index 26a313d8a..ab8850110 100644 --- a/repoze/bfg/router.py +++ b/repoze/bfg/router.py @@ -1,6 +1,4 @@ from zope.component import getAdapter -from zope.component import queryMultiAdapter -from zope.component import queryUtility from zope.component.event import dispatch from zope.interface import directlyProvides @@ -13,14 +11,16 @@ from repoze.bfg.events import NewResponse from repoze.bfg.events import WSGIApplicationCreatedEvent from repoze.bfg.interfaces import ITraverserFactory -from repoze.bfg.interfaces import IView -from repoze.bfg.interfaces import IViewPermission -from repoze.bfg.interfaces import ISecurityPolicy from repoze.bfg.interfaces import IRequest from repoze.bfg.registry import registry_manager from repoze.bfg.registry import makeRegistry +from repoze.bfg.security import Unauthorized + +from repoze.bfg.view import is_response +from repoze.bfg.view import render_view_to_response + _marker = () class Router: @@ -43,40 +43,27 @@ class Router: request.view_name = name request.subpath = subpath - security_policy = queryUtility(ISecurityPolicy) - if security_policy: - permission = queryMultiAdapter((context, request), IViewPermission, - name=name) - if permission is not None: - if not permission(security_policy): - app = HTTPUnauthorized() - app.explanation = repr(permission) - return app(environ, start_response) - - response = queryMultiAdapter((context, request), IView, name=name, - default=_marker) - if response is _marker: + try: + response = render_view_to_response(context, request, name, + secure=True) + except Unauthorized, why: + app = HTTPUnauthorized() + app.explanation = str(why) + return app(environ, start_response) + + if response is None: app = HTTPNotFound(request.url) return app(environ, start_response) - if not isResponse(response): - raise ValueError('response was not IResponse: %s' % response) + if not is_response(response): + raise ValueError('response did not implement IResponse: %r' + % response) dispatch(NewResponse(response)) start_response(response.status, response.headerlist) return response.app_iter -def isResponse(ob): - # response objects aren't obligated to implement a Zope interface, - # so we do it the hard way - if ( hasattr(ob, 'app_iter') and hasattr(ob, 'headerlist') and - hasattr(ob, 'status') ): - if ( hasattr(ob.app_iter, '__iter__') and - hasattr(ob.headerlist, '__iter__') and - isinstance(ob.status, basestring) ) : - return True - def make_app(root_policy, package=None, filename='configure.zcml', options=None): """ Create a view registry based on the application's ZCML. and diff --git a/repoze/bfg/security.py b/repoze/bfg/security.py index 6535f2608..226cf8664 100644 --- a/repoze/bfg/security.py +++ b/repoze/bfg/security.py @@ -203,7 +203,7 @@ class PermitsResult: self.context_repr = repr(context) def __str__(self): - msg = '%s: %r via ace %r in acl %s or principals %r in context %s' + msg = '%s: %r via ace %r in acl %r or principals %r in context %s' msg = msg % (self.__class__.__name__, self.permission, self.ace, self.acl, self.principals, self.context_repr) @@ -219,8 +219,7 @@ class Denied(PermitsResult): return False def __eq__(self, other): - if bool(other) is False: - return True + return bool(other) is False class Allowed(PermitsResult): """ An instance of ``Allowed`` is returned by an ACL allow. It @@ -232,8 +231,7 @@ class Allowed(PermitsResult): return True def __eq__(self, other): - if bool(other) is True: - return True + return bool(other) is True def flatten(x): """flatten(sequence) -> list @@ -281,4 +279,12 @@ class ViewPermissionFactory(object): def __call__(self, context, request): return ViewPermission(context, request, self.permission_name) +class Unauthorized(Exception): + def __init__(self, message='Unauthorized'): + self.message = message + + def __str__(self): + return str(self.message) + + diff --git a/repoze/bfg/tests/test_router.py b/repoze/bfg/tests/test_router.py index ed08710f1..a1c2b0cfb 100644 --- a/repoze/bfg/tests/test_router.py +++ b/repoze/bfg/tests/test_router.py @@ -154,7 +154,6 @@ class RouterTests(unittest.TestCase, PlacelessSetup): environ = self._makeEnviron() self._registerTraverserFactory(traversalfactory, '', None) self._registerView(view, '', IContext, IRequest) - app_context = make_appcontext() router = self._makeOne(rootpolicy, None) start_response = DummyStartResponse() result = router(environ, start_response) @@ -178,7 +177,6 @@ class RouterTests(unittest.TestCase, PlacelessSetup): self._registerView(view, '', IContext, IRequest) secpol = DummySecurityPolicy() self._registerSecurityPolicy(secpol) - app_context = make_appcontext() router = self._makeOne(rootpolicy, None) start_response = DummyStartResponse() result = router(environ, start_response) @@ -203,7 +201,6 @@ class RouterTests(unittest.TestCase, PlacelessSetup): self._registerView(view, '', IContext, IRequest) self._registerSecurityPolicy(secpol) self._registerPermission(permissionfactory, '', IContext, IRequest) - app_context = make_appcontext() router = self._makeOne(rootpolicy, None) start_response = DummyStartResponse() result = router(environ, start_response) @@ -223,13 +220,15 @@ class RouterTests(unittest.TestCase, PlacelessSetup): response = DummyResponse() view = make_view(response) secpol = DummySecurityPolicy() - permissionfactory = make_permission_factory(False) + from repoze.bfg.security import Denied + permissionfactory = make_permission_factory( + Denied('ace', 'acl', 'permission', ['principals'], context) + ) environ = self._makeEnviron() self._registerTraverserFactory(traversalfactory, '', None) self._registerView(view, '', IContext, IRequest) self._registerSecurityPolicy(secpol) self._registerPermission(permissionfactory, '', IContext, IRequest) - app_context = make_appcontext() router = self._makeOne(rootpolicy, None) start_response = DummyStartResponse() result = router(environ, start_response) @@ -364,13 +363,6 @@ def make_rootpolicy(root): return root return rootpolicy -def make_appcontext(): - from zope.configuration.interfaces import IConfigurationContext - from zope.interface import directlyProvides - app_context = DummyContext() - directlyProvides(app_context, IConfigurationContext) - return app_context - class DummyStartResponse: status = () headers = () diff --git a/repoze/bfg/tests/test_view.py b/repoze/bfg/tests/test_view.py new file mode 100644 index 000000000..3b776b14f --- /dev/null +++ b/repoze/bfg/tests/test_view.py @@ -0,0 +1,284 @@ +import unittest + +from zope.component.testing import PlacelessSetup + +class BaseTest(PlacelessSetup): + def setUp(self): + PlacelessSetup.setUp(self) + + def tearDown(self): + PlacelessSetup.tearDown(self) + + def _registerView(self, app, name, *for_): + import zope.component + gsm = zope.component.getGlobalSiteManager() + from repoze.bfg.interfaces import IView + gsm.registerAdapter(app, for_, IView, name) + + def _registerPermission(self, permission, name, *for_): + import zope.component + gsm = zope.component.getGlobalSiteManager() + from repoze.bfg.interfaces import IViewPermission + gsm.registerAdapter(permission, for_, IViewPermission, name) + + def _registerSecurityPolicy(self, secpol): + import zope.component + gsm = zope.component.getGlobalSiteManager() + from repoze.bfg.interfaces import ISecurityPolicy + gsm.registerUtility(secpol, ISecurityPolicy) + + def _makeEnviron(self, **extras): + environ = { + 'wsgi.url_scheme':'http', + 'SERVER_NAME':'localhost', + 'SERVER_PORT':'8080', + 'REQUEST_METHOD':'GET', + } + environ.update(extras) + return environ + +class RenderViewToResponseTests(unittest.TestCase, BaseTest): + def _getFUT(self): + from repoze.bfg.view import render_view_to_response + return render_view_to_response + + def test_call_no_view_registered(self): + environ = self._makeEnviron() + from webob import Request + request = Request(environ) + context = DummyContext() + renderer = self._getFUT() + result = renderer(context, request, name='notregistered') + self.assertEqual(result, None) + + def test_call_view_registered_secure_permission_disallows(self): + context = DummyContext() + from zope.interface import Interface + from zope.interface import directlyProvides + from repoze.bfg.interfaces import IRequest + class IContext(Interface): + pass + directlyProvides(context, IContext) + response = DummyResponse() + secpol = DummySecurityPolicy() + permissionfactory = make_permission_factory(False) + view = make_view(response) + self._registerView(view, 'registered', IContext, IRequest) + self._registerSecurityPolicy(secpol) + self._registerPermission(permissionfactory, 'registered', IContext, + IRequest) + environ = self._makeEnviron() + from webob import Request + request = Request(environ) + directlyProvides(request, IRequest) + renderer = self._getFUT() + from repoze.bfg.security import Unauthorized + self.assertRaises(Unauthorized, renderer, context, request, + name='registered', secure=True) + + def test_call_view_registered_secure_permission_allows(self): + context = DummyContext() + from zope.interface import Interface + from zope.interface import directlyProvides + from repoze.bfg.interfaces import IRequest + class IContext(Interface): + pass + directlyProvides(context, IContext) + response = DummyResponse() + secpol = DummySecurityPolicy() + permissionfactory = make_permission_factory(True) + view = make_view(response) + self._registerView(view, 'registered', IContext, IRequest) + self._registerSecurityPolicy(secpol) + self._registerPermission(permissionfactory, 'registered', IContext, + IRequest) + environ = self._makeEnviron() + from webob import Request + request = Request(environ) + directlyProvides(request, IRequest) + renderer = self._getFUT() + response = renderer(context, request, name='registered', secure=True) + self.assertEqual(response.status, '200 OK') + + def test_call_view_registered_insecure_permission_disallows(self): + context = DummyContext() + from zope.interface import Interface + from zope.interface import directlyProvides + from repoze.bfg.interfaces import IRequest + class IContext(Interface): + pass + directlyProvides(context, IContext) + response = DummyResponse() + secpol = DummySecurityPolicy() + permissionfactory = make_permission_factory(False) + view = make_view(response) + self._registerView(view, 'registered', IContext, IRequest) + self._registerSecurityPolicy(secpol) + self._registerPermission(permissionfactory, 'registered', IContext, + IRequest) + environ = self._makeEnviron() + from webob import Request + request = Request(environ) + directlyProvides(request, IRequest) + renderer = self._getFUT() + response = renderer(context, request, name='registered', secure=False) + self.assertEqual(response.status, '200 OK') + +class RenderViewToIterableTests(unittest.TestCase, BaseTest): + def _getFUT(self): + from repoze.bfg.view import render_view_to_iterable + return render_view_to_iterable + + def test_call_no_view_registered(self): + environ = self._makeEnviron() + from webob import Request + request = Request(environ) + context = DummyContext() + renderer = self._getFUT() + result = renderer(context, request, name='notregistered') + self.assertEqual(result, None) + + def test_call_view_registered_secure_permission_disallows(self): + context = DummyContext() + from zope.interface import Interface + from zope.interface import directlyProvides + from repoze.bfg.interfaces import IRequest + class IContext(Interface): + pass + directlyProvides(context, IContext) + response = DummyResponse() + secpol = DummySecurityPolicy() + permissionfactory = make_permission_factory(False) + view = make_view(response) + self._registerView(view, 'registered', IContext, IRequest) + self._registerSecurityPolicy(secpol) + self._registerPermission(permissionfactory, 'registered', IContext, + IRequest) + environ = self._makeEnviron() + from webob import Request + request = Request(environ) + directlyProvides(request, IRequest) + renderer = self._getFUT() + from repoze.bfg.security import Unauthorized + self.assertRaises(Unauthorized, renderer, context, request, + name='registered', secure=True) + + def test_call_view_registered_secure_permission_allows(self): + context = DummyContext() + from zope.interface import Interface + from zope.interface import directlyProvides + from repoze.bfg.interfaces import IRequest + class IContext(Interface): + pass + directlyProvides(context, IContext) + response = DummyResponse() + secpol = DummySecurityPolicy() + permissionfactory = make_permission_factory(True) + view = make_view(response) + self._registerView(view, 'registered', IContext, IRequest) + self._registerSecurityPolicy(secpol) + self._registerPermission(permissionfactory, 'registered', IContext, + IRequest) + environ = self._makeEnviron() + from webob import Request + request = Request(environ) + directlyProvides(request, IRequest) + renderer = self._getFUT() + iterable = renderer(context, request, name='registered', secure=True) + self.assertEqual(iterable, ()) + + def test_call_view_registered_insecure_permission_disallows(self): + context = DummyContext() + from zope.interface import Interface + from zope.interface import directlyProvides + from repoze.bfg.interfaces import IRequest + class IContext(Interface): + pass + directlyProvides(context, IContext) + response = DummyResponse() + secpol = DummySecurityPolicy() + permissionfactory = make_permission_factory(False) + view = make_view(response) + self._registerView(view, 'registered', IContext, IRequest) + self._registerSecurityPolicy(secpol) + self._registerPermission(permissionfactory, 'registered', IContext, + IRequest) + environ = self._makeEnviron() + from webob import Request + request = Request(environ) + directlyProvides(request, IRequest) + renderer = self._getFUT() + iterable = renderer(context, request, name='registered', secure=False) + self.assertEqual(iterable, ()) + + def test_call_view_response_doesnt_implement_IResponse(self): + context = DummyContext() + from zope.interface import Interface + from zope.interface import directlyProvides + from repoze.bfg.interfaces import IRequest + class IContext(Interface): + pass + directlyProvides(context, IContext) + response = 'abc' + view = make_view(response) + self._registerView(view, 'registered', IContext, IRequest) + environ = self._makeEnviron() + from webob import Request + request = Request(environ) + directlyProvides(request, IRequest) + renderer = self._getFUT() + self.assertRaises(ValueError, renderer, context, request, + name='registered', secure=False) + +class TestIsResponse(unittest.TestCase): + def _getFUT(self): + from repoze.bfg.view import is_response + return is_response + + def test_is(self): + response = DummyResponse() + f = self._getFUT() + self.assertEqual(f(response), True) + + def test_isnt(self): + response = None + f = self._getFUT() + self.assertEqual(f(response), False) + +class DummyContext: + pass + +def make_view(response): + def view(context, request): + return response + return view + +def make_permission_factory(result): + class DummyPermissionFactory: + def __init__(self, context, request): + self.context = context + self.request = request + + def __call__(self, secpol): + self.__class__.checked_with = secpol + return result + + def __repr__(self): + return 'permission' + return DummyPermissionFactory + +def make_appcontext(): + from zope.configuration.interfaces import IConfigurationContext + from zope.interface import directlyProvides + app_context = DummyContext() + directlyProvides(app_context, IConfigurationContext) + return app_context + +class DummyResponse: + status = '200 OK' + headerlist = () + app_iter = () + +class DummySecurityPolicy: + pass + diff --git a/repoze/bfg/view.py b/repoze/bfg/view.py new file mode 100644 index 000000000..180c020a0 --- /dev/null +++ b/repoze/bfg/view.py @@ -0,0 +1,71 @@ +from zope.component import queryMultiAdapter +from zope.component import queryUtility + +from repoze.bfg.interfaces import ISecurityPolicy +from repoze.bfg.interfaces import IViewPermission +from repoze.bfg.interfaces import IView +from repoze.bfg.security import Unauthorized + +def render_view_to_response(context, request, name='', secure=True): + """ Render the view named ``name`` against the specified + ``context`` and ``request`` to an object implementing + ``repoze.bfg.interfaces.IResponse`` or ``None`` if no such view + exists. This function will return ``None`` if a corresponding + view cannot be found. If ``secure`` is ``True``, and the view is + protected by a permission, the permission will be checked before + calling the view function. If the permission check disallows view + execution (based on the current security policy), a + ``repoze.bfg.security.Unauthorized`` exception will be raised; its + ``message`` attribute explains why the view access was disallowed. + If ``secure`` is ``False``, no permission checking is done.""" + if secure: + security_policy = queryUtility(ISecurityPolicy) + if security_policy: + permission = queryMultiAdapter((context, request), IViewPermission, + name=name) + if permission is not None: + result = permission(security_policy) + if not result: + raise Unauthorized(result) + return queryMultiAdapter((context, request), IView, name=name) + +def render_view_to_iterable(context, request, name='', secure=True): + """ Render the view named ``name`` against the specified + ``context`` and ``request``, and return an iterable representing + the view response's ``app_iter`` (see the interface named + ``repoze.bfg.interfaces.IResponse``). This function will return + ``None`` if a corresponding view cannot be found. Additionally, + this function will raise a ``ValueError`` if a view function is + found and called but the view does not return an object which + implements ``repoze.bfg.interfaces.IResponse``. You can usually + get the string representation of the return value of this function + by calling ``''.join(iterable)``. If ``secure`` is ``True``, and + the view is protected by a permission, the permission will be + checked before calling the view function. If the permission check + disallows view execution (based on the current security policy), a + ``repoze.bfg.security.Unauthorized`` exception will be raised; its + ``message`` attribute explains why the view access was disallowed. + If ``secure`` is ``False``, no permission checking is done.""" + response = render_view_to_response(context, request, name, secure) + if response is None: + return None + if not is_response(response): + raise ValueError('response did not implement IResponse: %r' % response) + return response.app_iter + +def is_response(ob): + """ Return True if ``ob`` implements the + ``repoze.bfg.interfaces.IResponse`` interface, False if not. Note + that this isn't actually a true Zope interface check, it's a + duck-typing check, as response objects are not obligated to + actually implement a Zope interface.""" + # response objects aren't obligated to implement a Zope interface, + # so we do it the hard way + if ( hasattr(ob, 'app_iter') and hasattr(ob, 'headerlist') and + hasattr(ob, 'status') ): + if ( hasattr(ob.app_iter, '__iter__') and + hasattr(ob.headerlist, '__iter__') and + isinstance(ob.status, basestring) ) : + return True + return False + |