From 7de404bb4af2744a64c13e31a780fc0229b8f9e5 Mon Sep 17 00:00:00 2001 From: Chris McDonough Date: Mon, 7 Jul 2008 04:44:57 +0000 Subject: Look up a view after traversal; adapt it to IWSGIApplication. --- repoze/bfg/tests/test_router.py | 263 ++++++++++++++++++++++------------------ 1 file changed, 145 insertions(+), 118 deletions(-) (limited to 'repoze/bfg/tests/test_router.py') diff --git a/repoze/bfg/tests/test_router.py b/repoze/bfg/tests/test_router.py index c684e06b5..bfa72fe41 100644 --- a/repoze/bfg/tests/test_router.py +++ b/repoze/bfg/tests/test_router.py @@ -9,12 +9,24 @@ class RouterTests(unittest.TestCase, PlacelessSetup): def tearDown(self): PlacelessSetup.tearDown(self) - def _registerFactory(self, app, name, *for_): + def _registerTraverserFactory(self, app, name, *for_): + import zope.component + gsm = zope.component.getGlobalSiteManager() + from repoze.bfg.interfaces import IPublishTraverserFactory + gsm.registerAdapter(app, for_, IPublishTraverserFactory, name) + + def _registerViewFactory(self, app, name, *for_): + import zope.component + gsm = zope.component.getGlobalSiteManager() + from repoze.bfg.interfaces import IViewFactory + gsm.registerAdapter(app, for_, IViewFactory, name) + + def _registerWSGIFactory(self, app, name, *for_): import zope.component gsm = zope.component.getGlobalSiteManager() from repoze.bfg.interfaces import IWSGIApplicationFactory gsm.registerAdapter(app, for_, IWSGIApplicationFactory, name) - + def _getTargetClass(self): from repoze.bfg.router import Router return Router @@ -33,151 +45,166 @@ class RouterTests(unittest.TestCase, PlacelessSetup): environ.update(extras) return environ - def test_call_no_app_registered(self): - statii = [] - headerii = [] - def rootpolicy(environ): - return None - def traversalpolicy(environ, root): - return DummyContext(), 'foo', [] - def securitypolicy(environ, context, name): - return None - def start_response(status, headers): - statii[:] = [status] - headerii[:] = [headers] + def test_call_no_view_registered(self): + rootpolicy = make_rootpolicy(None) environ = self._makeEnviron() - router = self._makeOne(rootpolicy, traversalpolicy, securitypolicy) + context = DummyContext() + traversalfactory = make_traversal_factory(context, '', []) + self._registerTraverserFactory(traversalfactory, '', None, None) + router = self._makeOne(rootpolicy) + start_response = DummyStartResponse() result = router(environ, start_response) - headers = headerii[0] + headers = start_response.headers self.assertEqual(len(headers), 2) - status = statii[0] + status = start_response.status self.assertEqual(status, '404 Not Found') self.failUnless('http://localhost:8080' in result[0], result) - def test_call_securitypolicy_denies(self): - statii = [] - headerii = [] - def rootpolicy(environ): - return None - def traversalpolicy(environ, root): - return DummyContext(), 'foo', [] - def securitypolicy(environ, context, name): - from webob.exc import HTTPUnauthorized - return HTTPUnauthorized() - def start_response(status, headers): - statii[:] = [status] - headerii[:] = [headers] - environ = self._makeEnviron() - router = self._makeOne(rootpolicy, traversalpolicy, securitypolicy) - result = router(environ, start_response) - headers = headerii[0] - self.assertEqual(len(headers), 2) - status = statii[0] - self.assertEqual(status, '401 Unauthorized') - - def test_call_app_registered_nonspecific_default_path(self): - def rootpolicy(environ): - return None + def test_call_view_registered_nonspecific_default_path(self): + rootpolicy = make_rootpolicy(None) context = DummyContext() - def traversalpolicy(environ, root): - return context, '', [] - def start_response(status, headers): - pass - def securitypolicy(environ, context, name): - return None + traversalfactory = make_traversal_factory(context, '', []) + response = DummyResponse() + viewfactory = make_view_factory(response) + wsgifactory = make_wsgi_factory('200 OK', (), ['Hello world']) environ = self._makeEnviron() - self._registerFactory(DummyWSGIApplicationFactory, '', None, None) - router = self._makeOne(rootpolicy, traversalpolicy, securitypolicy) + self._registerTraverserFactory(traversalfactory, '', None, None) + self._registerViewFactory(viewfactory, '', None, None) + self._registerWSGIFactory(wsgifactory, '', None, None) + router = self._makeOne(rootpolicy) + start_response = DummyStartResponse() 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'], []) - - def test_call_app_registered_nonspecific_nondefault_path_and_subpath(self): - def rootpolicy(environ): - return None + self.assertEqual(result, ['Hello world']) + self.assertEqual(start_response.headers, ()) + self.assertEqual(start_response.status, '200 OK') + request = environ['request'] + self.assertEqual(environ['request'].subpath, []) + self.assertEqual(environ['view'].context, context) + + def test_call_view_registered_nonspecific_nondefault_path_and_subpath(self): + rootpolicy = make_rootpolicy(None) context = DummyContext() - def traversalpolicy(environ, root): - return context, 'foo', ['bar', 'baz'] - def start_response(status, headers): - pass - def securitypolicy(environ, context, name): - return None + traversalfactory = make_traversal_factory(context, 'foo', ['bar']) + response = DummyResponse() + viewfactory = make_view_factory(response) + wsgifactory = make_wsgi_factory('200 OK', (), ['Hello world']) environ = self._makeEnviron() - self._registerFactory(DummyWSGIApplicationFactory, 'foo', None, None) - router = self._makeOne(rootpolicy, traversalpolicy, securitypolicy) + self._registerTraverserFactory(traversalfactory, '', None, None) + self._registerViewFactory(viewfactory, 'foo', None, None) + self._registerWSGIFactory(wsgifactory, '', None, None) + router = self._makeOne(rootpolicy) + start_response = DummyStartResponse() 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_success(self): - def rootpolicy(environ): - return None - context = DummyContext() + self.assertEqual(result, ['Hello world']) + self.assertEqual(start_response.headers, ()) + self.assertEqual(start_response.status, '200 OK') + request = environ['request'] + self.assertEqual(environ['request'].subpath, ['bar']) + self.assertEqual(environ['view'].context, context) + + def test_call_view_registered_specific_success(self): + rootpolicy = make_rootpolicy(None) from zope.interface import Interface from zope.interface import directlyProvides class IContext(Interface): pass + from repoze.bfg.interfaces import IRequest + context = DummyContext() directlyProvides(context, IContext) - def traversalpolicy(environ, root): - return context, 'foo', ['bar', 'baz'] - def start_response(status, headers): - pass - def securitypolicy(environ, context, name): - return None + traversalfactory = make_traversal_factory(context, '', []) + response = DummyResponse() + viewfactory = make_view_factory(response) + wsgifactory = make_wsgi_factory('200 OK', (), ['Hello world']) environ = self._makeEnviron() - from repoze.bfg.interfaces import IWebObRequest - self._registerFactory(DummyWSGIApplicationFactory, 'foo', IContext, - IWebObRequest) - router = self._makeOne(rootpolicy, traversalpolicy, securitypolicy) + self._registerTraverserFactory(traversalfactory, '', None, None) + self._registerViewFactory(viewfactory, '', IContext, IRequest) + self._registerWSGIFactory(wsgifactory, '', None, None) + router = self._makeOne(rootpolicy) + start_response = DummyStartResponse() 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']) + self.assertEqual(result, ['Hello world']) + self.assertEqual(start_response.headers, ()) + self.assertEqual(start_response.status, '200 OK') + request = environ['request'] + self.assertEqual(environ['request'].subpath, []) + self.assertEqual(environ['view'].context, context) - def test_call_app_registered_specific_fail(self): - context = DummyContext() + def test_call_view_registered_specific_fail(self): + rootpolicy = make_rootpolicy(None) from zope.interface import Interface from zope.interface import directlyProvides - class INotContext(Interface): - pass class IContext(Interface): pass + class INotContext(Interface): + pass + from repoze.bfg.interfaces import IRequest + context = DummyContext() directlyProvides(context, INotContext) - statii = [] - headerii = [] - def rootpolicy(environ): - return None - def traversalpolicy(environ, root): - return context, 'foo', [] - def start_response(status, headers): - statii[:] = [status] - headerii[:] = [headers] - def securitypolicy(environ, context, name): - return None + traversalfactory = make_traversal_factory(context, '', ['']) + response = DummyResponse() + viewfactory = make_view_factory(response) + wsgifactory = make_wsgi_factory('200 OK', (), ['Hello world']) environ = self._makeEnviron() - from repoze.bfg.interfaces import IWebObRequest - self._registerFactory(DummyWSGIApplicationFactory, 'foo', IContext, - IWebObRequest) - router = self._makeOne(rootpolicy, traversalpolicy, securitypolicy) + self._registerTraverserFactory(traversalfactory, '', None, None) + self._registerViewFactory(viewfactory, '', IContext, IRequest) + self._registerWSGIFactory(wsgifactory, '', None, None) + router = self._makeOne(rootpolicy) + start_response = DummyStartResponse() 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) + self.failUnless('404' in result[0]) + self.assertEqual(start_response.status, '404 Not Found') class DummyContext: pass -class DummyWSGIApplicationFactory: - def __init__(self, context, request): - self.context = context - self.request = request +def make_wsgi_factory(status, headers, app_iter): + class DummyWSGIApplicationFactory: + def __init__(self, view, request): + self.view = view + self.request = request + + def __call__(self, environ, start_response): + environ['view'] = self.view + environ['request'] = self.request + start_response(status, headers) + return app_iter + + return DummyWSGIApplicationFactory + +def make_view_factory(response): + class DummyViewFactory: + def __init__(self, context, request): + self.context = context + self.request = request + + def __call__(self): + return response + return DummyViewFactory - def __call__(self, environ, start_response): - return self.context, self.request +def make_traversal_factory(context, name, subpath): + class DummyTraversalFactory: + def __init__(self, root, request): + self.root = root + self.request = request + + def __call__(self, path): + return context, name, subpath + return DummyTraversalFactory + +def make_rootpolicy(root): + def rootpolicy(environ): + return root + return rootpolicy + +class DummyStartResponse: + status = () + headers = () + def __call__(self, status, headers): + self.status = status + self.headers = headers + +class DummyResponse: + status = '200 OK' + headerlist = () + app_iter = () + -- cgit v1.2.3