From 79957510964157dae99201b1b6a14ef209bc50f7 Mon Sep 17 00:00:00 2001 From: Chris McDonough Date: Sat, 5 Jul 2008 02:29:54 +0000 Subject: Use WebOb. Return a NotFound when we can't adapt. --- repoze/bfg/interfaces.py | 3 + repoze/bfg/router.py | 15 ++++- repoze/bfg/tests/test_router.py | 125 ++++++++++++++++++++++++++++++---------- setup.py | 8 ++- 4 files changed, 118 insertions(+), 33 deletions(-) diff --git a/repoze/bfg/interfaces.py b/repoze/bfg/interfaces.py index e915519e0..803cf237a 100644 --- a/repoze/bfg/interfaces.py +++ b/repoze/bfg/interfaces.py @@ -19,3 +19,6 @@ class ITraverser(Interface): def __call__(environ, name): """ Return a subcontext or based on name """ +class IWebObRequest(Interface): + """ Marker interface for a webob.Request object """ + diff --git a/repoze/bfg/router.py b/repoze/bfg/router.py index f37ca0f7c..626080bdc 100644 --- a/repoze/bfg/router.py +++ b/repoze/bfg/router.py @@ -1,5 +1,11 @@ -from zope.component import getAdapter +from zope.component import queryMultiAdapter +from zope.interface import directlyProvides + +from webob import Request +from webob.exc import HTTPNotFound + from repoze.bfg.interfaces import IWSGIApplicationFactory +from repoze.bfg.interfaces import IWebObRequest class Router: def __init__(self, root_policy, traversal_policy): @@ -10,5 +16,10 @@ class Router: root = self.root_policy(environ) context, name, subpath = self.traversal_policy(root, environ) environ['repoze.bfg.subpath'] = subpath - app = getAdapter(context, IWSGIApplicationFactory, name=name) + request = Request(environ) + directlyProvides(request, IWebObRequest) + app = queryMultiAdapter((context, request), + IWSGIApplicationFactory, name=name) + if app is None: + app = HTTPNotFound(request.url) return app(environ, start_response) diff --git a/repoze/bfg/tests/test_router.py b/repoze/bfg/tests/test_router.py index 46fdecbbc..8f4b2ff95 100644 --- a/repoze/bfg/tests/test_router.py +++ b/repoze/bfg/tests/test_router.py @@ -9,11 +9,11 @@ class RouterTests(unittest.TestCase, PlacelessSetup): def tearDown(self): PlacelessSetup.tearDown(self) - def _registerFactory(self, app, for_, name): + def _registerFactory(self, app, name, *for_): import zope.component gsm = zope.component.getGlobalSiteManager() from repoze.bfg.interfaces import IWSGIApplicationFactory - gsm.registerAdapter(app, (for_,), IWSGIApplicationFactory, name) + gsm.registerAdapter(app, for_, IWSGIApplicationFactory, name) def _getTargetClass(self): from repoze.bfg.router import Router @@ -23,63 +23,130 @@ class RouterTests(unittest.TestCase, PlacelessSetup): klass = self._getTargetClass() return klass(*arg, **kw) + def _makeEnviron(self, **extras): + environ = { + 'wsgi.url_scheme':'http', + 'SERVER_NAME':'localhost', + 'SERVER_PORT':'8080', + 'REQUEST_METHOD':'GET', + } + environ.update(extras) + return environ + def test_call_no_app_registered(self): + statii = [] + headerii = [] def rootpolicy(environ): return None def traversalpolicy(root, environ): return DummyContext(), 'foo', [] def start_response(status, headers): - pass - environ = {} + statii[:] = [status] + headerii[:] = [headers] + environ = self._makeEnviron() router = self._makeOne(rootpolicy, traversalpolicy) - from zope.component import ComponentLookupError - self.assertRaises(ComponentLookupError, router, environ, start_response) + result = router(environ, start_response) + headers = headerii[0] + self.assertEqual(len(headers), 2) + status = statii[0] + self.assertEqual(status, '404 Not Found') + self.failUnless('http://localhost:8080' in result[0], result) - def test_call_app_registered_default_path(self): + def test_call_app_registered_nonspecific_default_path(self): def rootpolicy(environ): return None context = DummyContext() - _marker = [] def traversalpolicy(root, environ): return context, '', [] def start_response(status, headers): pass - class DummyWSGIApplicationFactory: - def __init__(self, context): - self.context = context - - def __call__(self, environ, start_response): - return _marker - environ = {} - self._registerFactory(DummyWSGIApplicationFactory, None, '') + environ = self._makeEnviron() + self._registerFactory(DummyWSGIApplicationFactory, '', None, None) router = self._makeOne(rootpolicy, traversalpolicy) result = router(environ, start_response) - self.failUnless(result is _marker) + self.failUnless(result[0] is context) + import webob + self.failUnless(isinstance(result[1], webob.Request)) self.assertEqual(environ['repoze.bfg.subpath'], []) - def test_call_app_registered_nondefault_path_and_subpath(self): + def test_call_app_registered_nonspecific_nondefault_path_and_subpath(self): def rootpolicy(environ): return None context = DummyContext() - _marker = [] def traversalpolicy(root, environ): return context, 'foo', ['bar', 'baz'] def start_response(status, headers): pass - class DummyWSGIApplicationFactory: - def __init__(self, context): - self.context = context + environ = self._makeEnviron() + self._registerFactory(DummyWSGIApplicationFactory, 'foo', None, None) + router = self._makeOne(rootpolicy, traversalpolicy) + result = router(environ, start_response) + self.failUnless(result[0] is context) + import webob + self.failUnless(isinstance(result[1], webob.Request)) + self.assertEqual(environ['repoze.bfg.subpath'], ['bar', 'baz']) - def __call__(self, environ, start_response): - return _marker - environ = {} - self._registerFactory(DummyWSGIApplicationFactory, None, 'foo') + def test_call_app_registered_specific_success(self): + def rootpolicy(environ): + return None + context = DummyContext() + from zope.interface import Interface + from zope.interface import directlyProvides + class IContext(Interface): + pass + directlyProvides(context, IContext) + def traversalpolicy(root, environ): + return context, 'foo', ['bar', 'baz'] + def start_response(status, headers): + pass + environ = self._makeEnviron() + from repoze.bfg.interfaces import IWebObRequest + self._registerFactory(DummyWSGIApplicationFactory, 'foo', IContext, + IWebObRequest) router = self._makeOne(rootpolicy, traversalpolicy) result = router(environ, start_response) - self.failUnless(result is _marker) + self.failUnless(result[0] is context) + import webob + self.failUnless(isinstance(result[1], webob.Request)) self.assertEqual(environ['repoze.bfg.subpath'], ['bar', 'baz']) - + + def test_call_app_registered_specific_fail(self): + context = DummyContext() + from zope.interface import Interface + from zope.interface import directlyProvides + class INotContext(Interface): + pass + class IContext(Interface): + pass + directlyProvides(context, INotContext) + statii = [] + headerii = [] + def rootpolicy(environ): + return None + def traversalpolicy(root, environ): + return context, 'foo', [] + def start_response(status, headers): + statii[:] = [status] + headerii[:] = [headers] + environ = self._makeEnviron() + from repoze.bfg.interfaces import IWebObRequest + self._registerFactory(DummyWSGIApplicationFactory, 'foo', IContext, + IWebObRequest) + router = self._makeOne(rootpolicy, traversalpolicy) + result = router(environ, start_response) + headers = headerii[0] + self.assertEqual(len(headers), 2) + status = statii[0] + self.assertEqual(status, '404 Not Found') + self.failUnless('http://localhost:8080' in result[0], result) + class DummyContext: pass - +class DummyWSGIApplicationFactory: + def __init__(self, context, request): + self.context = context + self.request = request + + def __call__(self, environ, start_response): + return self.context, self.request diff --git a/setup.py b/setup.py index 8015615ff..dce4b7ceb 100644 --- a/setup.py +++ b/setup.py @@ -50,11 +50,15 @@ setup(name='repoze.bfg', install_requires=[ 'zope.interface', 'zope.component', - 'zope.testing'], + 'zope.testing', + 'WebOb' + ], tests_require=[ 'zope.interface', 'zope.component', - 'zope.testing'], + 'zope.testing', + 'WebOb', + ], test_suite="repoze.bfg.tests", entry_points = """\ """ -- cgit v1.2.3