[Repoze-checkins] r1234 - in repoze.bfg/trunk: . repoze/bfg repoze/bfg/tests
Chris McDonough
chrism at agendaless.com
Fri Jul 4 22:29:54 EDT 2008
Author: Chris McDonough <chrism at agendaless.com>
Date: Fri Jul 4 22:29:54 2008
New Revision: 1234
Log:
Use WebOb.
Return a NotFound when we can't adapt.
Modified:
repoze.bfg/trunk/repoze/bfg/interfaces.py
repoze.bfg/trunk/repoze/bfg/router.py
repoze.bfg/trunk/repoze/bfg/tests/test_router.py
repoze.bfg/trunk/setup.py
Modified: repoze.bfg/trunk/repoze/bfg/interfaces.py
==============================================================================
--- repoze.bfg/trunk/repoze/bfg/interfaces.py (original)
+++ repoze.bfg/trunk/repoze/bfg/interfaces.py Fri Jul 4 22:29:54 2008
@@ -19,3 +19,6 @@
def __call__(environ, name):
""" Return a subcontext or based on name """
+class IWebObRequest(Interface):
+ """ Marker interface for a webob.Request object """
+
Modified: repoze.bfg/trunk/repoze/bfg/router.py
==============================================================================
--- repoze.bfg/trunk/repoze/bfg/router.py (original)
+++ repoze.bfg/trunk/repoze/bfg/router.py Fri Jul 4 22:29:54 2008
@@ -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 @@
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)
Modified: repoze.bfg/trunk/repoze/bfg/tests/test_router.py
==============================================================================
--- repoze.bfg/trunk/repoze/bfg/tests/test_router.py (original)
+++ repoze.bfg/trunk/repoze/bfg/tests/test_router.py Fri Jul 4 22:29:54 2008
@@ -9,11 +9,11 @@
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 @@
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
-
- def __call__(self, environ, start_response):
- return _marker
- environ = {}
- self._registerFactory(DummyWSGIApplicationFactory, None, 'foo')
+ environ = self._makeEnviron()
+ self._registerFactory(DummyWSGIApplicationFactory, 'foo', 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'], ['bar', 'baz'])
-
+
+ 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[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
Modified: repoze.bfg/trunk/setup.py
==============================================================================
--- repoze.bfg/trunk/setup.py (original)
+++ repoze.bfg/trunk/setup.py Fri Jul 4 22:29:54 2008
@@ -50,11 +50,15 @@
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 = """\
"""
More information about the Repoze-checkins
mailing list