1
1
"""Test the various means of instantiating and invoking tools."""
6
from httplib import IncompleteRead
5
from cherrypy._cpcompat import BytesIO, copyitems, itervalues
6
from cherrypy._cpcompat import IncompleteRead, ntob, ntou, py3k, xrange
11
from cherrypy.test import test
12
test.prefer_parent_path()
15
12
from cherrypy import tools
18
europoundUnicode = u'\x80\xa3'
22
# Put check_access in a custom toolbox with its own namespace
23
myauthtools = cherrypy._cptools.Toolbox("myauth")
25
def check_access(default=False):
26
if not getattr(cherrypy.request, "userid", default):
27
raise cherrypy.HTTPError(401)
28
myauthtools.check_access = cherrypy.Tool('before_request_body', check_access)
33
for k, v in cherrypy.request.numerify_map:
34
chunk = chunk.replace(k, v)
36
cherrypy.response.body = number_it(cherrypy.response.body)
38
class NumTool(cherrypy.Tool):
41
m = self._merged_args().get("map", {})
42
cherrypy.request.numerify_map = m.items()
43
cherrypy.request.hooks.attach('on_start_resource', makemap)
46
cherrypy.request.error_response = cherrypy.HTTPError(502).set_response
47
critical.failsafe = True
49
cherrypy.request.hooks.attach('on_start_resource', critical)
50
cherrypy.request.hooks.attach(self._point, self.callable)
52
tools.numerify = NumTool('before_finalize', numerify)
54
# It's not mandatory to inherit from cherrypy.Tool.
62
def nadsat_it_up(body):
64
chunk = chunk.replace("good", "horrorshow")
65
chunk = chunk.replace("piece", "lomtick")
67
cherrypy.response.body = nadsat_it_up(cherrypy.response.body)
71
# This runs after the request has been completely written out.
72
cherrypy.response.body = "razdrez"
73
id = cherrypy.request.params.get("id")
76
cleanup.failsafe = True
79
cherrypy.request.hooks.attach('before_finalize', self.nadsat)
80
cherrypy.request.hooks.attach('on_end_request', self.cleanup)
81
tools.nadsat = NadsatTool()
84
cherrypy.request.process_request_body = False
85
clen = int(cherrypy.request.headers['Content-Length'])
86
cherrypy.request.body = cherrypy.request.rfile.read(clen)
88
# Assert that we can use a callable object instead of a function.
89
class Rotator(object):
90
def __call__(self, scale):
93
r.body = [chr(ord(x) + scale) for x in r.body]
94
cherrypy.tools.rotator = cherrypy.Tool('before_finalize', Rotator())
96
def stream_handler(next_handler, *args, **kwargs):
97
cherrypy.response.output = o = StringIO.StringIO()
99
response = next_handler(*args, **kwargs)
100
# Ignore the response and return our accumulated output instead.
104
cherrypy.tools.streamer = cherrypy._cptools.HandlerWrapperTool(stream_handler)
108
return "Howdy earth!"
112
cherrypy.response.output.write('I am ')
113
cherrypy.response.output.write('a tarfile')
114
tarfile.exposed = True
115
tarfile._cp_config = {'tools.streamer.on': True}
118
hooks = list(cherrypy.request.hooks['before_finalize'])
120
assert [x.callback.__name__ for x in hooks] == ['encode', 'gzip']
121
assert [x.priority for x in hooks] == [70, 80]
124
yield europoundUnicode
129
return cherrypy.request.body
131
pipe._cp_config = {'hooks.before_request_body': pipe_body}
133
# Multiple decorators; include kwargs just for fun.
134
# Note that encode must run before gzip.
135
def decorated_euro(self, *vpath):
138
yield europoundUnicode
139
decorated_euro.exposed = True
140
decorated_euro = tools.gzip(compress_level=6)(decorated_euro)
141
decorated_euro = tools.encode(errors='ignore')(decorated_euro)
146
class TestType(type):
147
"""Metaclass which automatically exposes all functions in each subclass,
148
and adds an instance of the subclass as an attribute of root.
150
def __init__(cls, name, bases, dct):
151
type.__init__(cls, name, bases, dct)
152
for value in dct.itervalues():
153
if isinstance(value, types.FunctionType):
155
setattr(root, name.lower(), cls())
157
__metaclass__ = TestType
161
# Declare Tools in _cp_config
164
_cp_config = {"tools.nadsat.on": True}
166
def index(self, id=None):
167
return "A good piece of cherry pie"
170
return repr(tools.nadsat.ended[id])
172
def err(self, id=None):
175
def errinstream(self, id=None):
176
yield "nonconfidential"
180
# METHOD TWO: decorator using Tool()
181
# We support Python 2.3, but the @-deco syntax would look like this:
182
# @tools.check_access()
183
def restricted(self):
185
restricted = myauthtools.check_access()(restricted)
188
def err_in_onstart(self):
191
def stream(self, id=None):
192
for x in xrange(100000000):
194
stream._cp_config = {'response.stream': True}
197
cherrypy.config.update({'environment': 'test_suite'})
201
# Declare Tools in detached config
203
'tools.numerify.on': True,
204
'tools.numerify.map': {"pie": "3.14159"},
206
'/demo/restricted': {
207
'request.show_tracebacks': False,
210
'request.show_tracebacks': False,
211
'myauth.check_access.default': True,
213
'/demo/errinstream': {
214
'response.stream': True,
216
'/demo/err_in_onstart': {
217
# Because this isn't a dict, on_start_resource will error.
218
'tools.numerify.map': "pie->3.14159"
222
'tools.gzip.on': True,
223
'tools.encode.on': True,
225
# Priority specified in config
226
'/decorated_euro/subpath': {
227
'tools.gzip.priority': 10,
230
'/tarfile': {'tools.streamer.on': True}
232
app = cherrypy.tree.mount(root, config=conf)
233
app.request_class.namespaces['myauth'] = myauthtools
235
if sys.version_info >= (2, 5):
236
from cherrypy.test import py25
237
root.tooldecs = py25.ToolExamples()
15
europoundUnicode = ntou('\x80\xa3')
240
18
# Client-side code #
245
23
class ToolTests(helper.CPWebCase):
26
# Put check_access in a custom toolbox with its own namespace
27
myauthtools = cherrypy._cptools.Toolbox("myauth")
29
def check_access(default=False):
30
if not getattr(cherrypy.request, "userid", default):
31
raise cherrypy.HTTPError(401)
32
myauthtools.check_access = cherrypy.Tool('before_request_body', check_access)
37
for k, v in cherrypy.request.numerify_map:
38
chunk = chunk.replace(k, v)
40
cherrypy.response.body = number_it(cherrypy.response.body)
42
class NumTool(cherrypy.Tool):
45
m = self._merged_args().get("map", {})
46
cherrypy.request.numerify_map = copyitems(m)
47
cherrypy.request.hooks.attach('on_start_resource', makemap)
50
cherrypy.request.error_response = cherrypy.HTTPError(502).set_response
51
critical.failsafe = True
53
cherrypy.request.hooks.attach('on_start_resource', critical)
54
cherrypy.request.hooks.attach(self._point, self.callable)
56
tools.numerify = NumTool('before_finalize', numerify)
58
# It's not mandatory to inherit from cherrypy.Tool.
66
def nadsat_it_up(body):
68
chunk = chunk.replace(ntob("good"), ntob("horrorshow"))
69
chunk = chunk.replace(ntob("piece"), ntob("lomtick"))
71
cherrypy.response.body = nadsat_it_up(cherrypy.response.body)
75
# This runs after the request has been completely written out.
76
cherrypy.response.body = [ntob("razdrez")]
77
id = cherrypy.request.params.get("id")
80
cleanup.failsafe = True
83
cherrypy.request.hooks.attach('before_finalize', self.nadsat)
84
cherrypy.request.hooks.attach('on_end_request', self.cleanup)
85
tools.nadsat = NadsatTool()
88
cherrypy.request.process_request_body = False
89
clen = int(cherrypy.request.headers['Content-Length'])
90
cherrypy.request.body = cherrypy.request.rfile.read(clen)
92
# Assert that we can use a callable object instead of a function.
93
class Rotator(object):
94
def __call__(self, scale):
98
r.body = [bytes([(x + scale) % 256 for x in r.body[0]])]
100
r.body = [chr((ord(x) + scale) % 256) for x in r.body[0]]
101
cherrypy.tools.rotator = cherrypy.Tool('before_finalize', Rotator())
103
def stream_handler(next_handler, *args, **kwargs):
104
cherrypy.response.output = o = BytesIO()
106
response = next_handler(*args, **kwargs)
107
# Ignore the response and return our accumulated output instead.
111
cherrypy.tools.streamer = cherrypy._cptools.HandlerWrapperTool(stream_handler)
115
return "Howdy earth!"
119
cherrypy.response.output.write(ntob('I am '))
120
cherrypy.response.output.write(ntob('a tarfile'))
121
tarfile.exposed = True
122
tarfile._cp_config = {'tools.streamer.on': True}
125
hooks = list(cherrypy.request.hooks['before_finalize'])
127
cbnames = [x.callback.__name__ for x in hooks]
128
assert cbnames == ['gzip'], cbnames
129
priorities = [x.priority for x in hooks]
130
assert priorities == [80], priorities
133
yield europoundUnicode
138
return cherrypy.request.body
140
pipe._cp_config = {'hooks.before_request_body': pipe_body}
142
# Multiple decorators; include kwargs just for fun.
143
# Note that rotator must run before gzip.
144
def decorated_euro(self, *vpath):
147
yield europoundUnicode
148
decorated_euro.exposed = True
149
decorated_euro = tools.gzip(compress_level=6)(decorated_euro)
150
decorated_euro = tools.rotator(scale=3)(decorated_euro)
155
class TestType(type):
156
"""Metaclass which automatically exposes all functions in each subclass,
157
and adds an instance of the subclass as an attribute of root.
159
def __init__(cls, name, bases, dct):
160
type.__init__(cls, name, bases, dct)
161
for value in itervalues(dct):
162
if isinstance(value, types.FunctionType):
164
setattr(root, name.lower(), cls())
165
Test = TestType('Test', (object,), {})
169
# Declare Tools in _cp_config
172
_cp_config = {"tools.nadsat.on": True}
174
def index(self, id=None):
175
return "A good piece of cherry pie"
178
return repr(tools.nadsat.ended[id])
180
def err(self, id=None):
183
def errinstream(self, id=None):
184
yield "nonconfidential"
188
# METHOD TWO: decorator using Tool()
189
# We support Python 2.3, but the @-deco syntax would look like this:
190
# @tools.check_access()
191
def restricted(self):
193
restricted = myauthtools.check_access()(restricted)
196
def err_in_onstart(self):
199
def stream(self, id=None):
200
for x in xrange(100000000):
202
stream._cp_config = {'response.stream': True}
207
# Declare Tools in detached config
209
'tools.numerify.on': True,
210
'tools.numerify.map': {ntob("pie"): ntob("3.14159")},
212
'/demo/restricted': {
213
'request.show_tracebacks': False,
216
'request.show_tracebacks': False,
217
'myauth.check_access.default': True,
219
'/demo/errinstream': {
220
'response.stream': True,
222
'/demo/err_in_onstart': {
223
# Because this isn't a dict, on_start_resource will error.
224
'tools.numerify.map': "pie->3.14159"
228
'tools.gzip.on': True,
229
'tools.encode.on': True,
231
# Priority specified in config
232
'/decorated_euro/subpath': {
233
'tools.gzip.priority': 10,
236
'/tarfile': {'tools.streamer.on': True}
238
app = cherrypy.tree.mount(root, config=conf)
239
app.request_class.namespaces['myauth'] = myauthtools
241
if sys.version_info >= (2, 5):
242
from cherrypy.test import _test_decorators
243
root.tooldecs = _test_decorators.ToolExamples()
244
setup_server = staticmethod(setup_server)
247
246
def testHookErrors(self):
248
247
self.getPage("/demo/?id=1")
249
248
# If body is "razdrez", then on_end_request is being called too early.
352
350
self.getPage("/decorated_euro", headers=[("Accept-Encoding", "gzip")])
353
351
self.assertInBody(zbuf.getvalue()[:3])
355
# This should break because gzip's priority was lowered in conf.
353
# This returns a different value because gzip's priority was
354
# lowered in conf, allowing the rotator to run after gzip.
356
355
# Of course, we don't want breakage in production apps,
357
356
# but it proves the priority was changed.
358
357
self.getPage("/decorated_euro/subpath",
359
358
headers=[("Accept-Encoding", "gzip")])
360
self.assertErrorPage(500, pattern='UnicodeEncodeError')
360
self.assertInBody(bytes([(x + 3) % 256 for x in zbuf.getvalue()]))
362
self.assertInBody(''.join([chr((ord(x) + 3) % 256) for x in zbuf.getvalue()]))
362
364
def testBareHooks(self):
363
365
content = "bit of a pain in me gulliver"
364
366
self.getPage("/pipe",
365
headers=[("Content-Length", len(content)),
367
headers=[("Content-Length", str(len(content))),
366
368
("Content-Type", "text/plain")],
367
369
method="POST", body=content)
368
370
self.assertBody(content)