1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
|
import unittest
from zope.component.testing import PlacelessSetup
class RouterTests(unittest.TestCase, PlacelessSetup):
def setUp(self):
PlacelessSetup.setUp(self)
def tearDown(self):
PlacelessSetup.tearDown(self)
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
def _makeOne(self, *arg, **kw):
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_view_registered(self):
rootpolicy = make_rootpolicy(None)
environ = self._makeEnviron()
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 = start_response.headers
self.assertEqual(len(headers), 2)
status = start_response.status
self.assertEqual(status, '404 Not Found')
self.failUnless('http://localhost:8080' in result[0], result)
def test_call_view_registered_nonspecific_default_path(self):
rootpolicy = make_rootpolicy(None)
context = DummyContext()
traversalfactory = make_traversal_factory(context, '', [])
response = DummyResponse()
viewfactory = make_view_factory(response)
wsgifactory = make_wsgi_factory('200 OK', (), ['Hello world'])
environ = self._makeEnviron()
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.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()
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._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.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)
traversalfactory = make_traversal_factory(context, '', [])
response = DummyResponse()
viewfactory = make_view_factory(response)
wsgifactory = make_wsgi_factory('200 OK', (), ['Hello world'])
environ = self._makeEnviron()
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.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_specific_fail(self):
rootpolicy = make_rootpolicy(None)
from zope.interface import Interface
from zope.interface import directlyProvides
class IContext(Interface):
pass
class INotContext(Interface):
pass
from repoze.bfg.interfaces import IRequest
context = DummyContext()
directlyProvides(context, INotContext)
traversalfactory = make_traversal_factory(context, '', [''])
response = DummyResponse()
viewfactory = make_view_factory(response)
wsgifactory = make_wsgi_factory('200 OK', (), ['Hello world'])
environ = self._makeEnviron()
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('404' in result[0])
self.assertEqual(start_response.status, '404 Not Found')
class DummyContext:
pass
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 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 = ()
|
