Package cherrypy :: Package test :: Module test_tools
[hide private]
[frames] | no frames]

Source Code for Module cherrypy.test.test_tools

  1  """Test the various means of instantiating and invoking tools.""" 
  2   
  3  import gzip 
  4  import StringIO 
  5  import sys 
  6  from httplib import IncompleteRead 
  7  import time 
  8  timeout = 0.2 
  9   
 10  import types 
 11  from cherrypy.test import test 
 12  test.prefer_parent_path() 
 13   
 14  import cherrypy 
 15  from cherrypy import tools 
 16   
 17   
 18  europoundUnicode = u'\x80\xa3' 
 19   
20 -def setup_server():
21 22 # Put check_access in a custom toolbox with its own namespace 23 myauthtools = cherrypy._cptools.Toolbox("myauth") 24 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) 29 30 def numerify(): 31 def number_it(body): 32 for chunk in body: 33 for k, v in cherrypy.request.numerify_map: 34 chunk = chunk.replace(k, v) 35 yield chunk 36 cherrypy.response.body = number_it(cherrypy.response.body) 37 38 class NumTool(cherrypy.Tool): 39 def _setup(self): 40 def makemap(): 41 m = self._merged_args().get("map", {}) 42 cherrypy.request.numerify_map = m.items() 43 cherrypy.request.hooks.attach('on_start_resource', makemap) 44 45 def critical(): 46 cherrypy.request.error_response = cherrypy.HTTPError(502).set_response 47 critical.failsafe = True 48 49 cherrypy.request.hooks.attach('on_start_resource', critical) 50 cherrypy.request.hooks.attach(self._point, self.callable) 51 52 tools.numerify = NumTool('before_finalize', numerify) 53 54 # It's not mandatory to inherit from cherrypy.Tool. 55 class NadsatTool: 56 57 def __init__(self): 58 self.ended = {} 59 self._name = "nadsat" 60 61 def nadsat(self): 62 def nadsat_it_up(body): 63 for chunk in body: 64 chunk = chunk.replace("good", "horrorshow") 65 chunk = chunk.replace("piece", "lomtick") 66 yield chunk 67 cherrypy.response.body = nadsat_it_up(cherrypy.response.body) 68 nadsat.priority = 0 69 70 def cleanup(self): 71 # This runs after the request has been completely written out. 72 cherrypy.response.body = "razdrez" 73 id = cherrypy.request.params.get("id") 74 if id: 75 self.ended[id] = True 76 cleanup.failsafe = True 77 78 def _setup(self): 79 cherrypy.request.hooks.attach('before_finalize', self.nadsat) 80 cherrypy.request.hooks.attach('on_end_request', self.cleanup) 81 tools.nadsat = NadsatTool() 82 83 def pipe_body(): 84 cherrypy.request.process_request_body = False 85 clen = int(cherrypy.request.headers['Content-Length']) 86 cherrypy.request.body = cherrypy.request.rfile.read(clen) 87 88 # Assert that we can use a callable object instead of a function. 89 class Rotator(object): 90 def __call__(self, scale): 91 r = cherrypy.response 92 r.collapse_body() 93 r.body = [chr(ord(x) + scale) for x in r.body] 94 cherrypy.tools.rotator = cherrypy.Tool('before_finalize', Rotator()) 95 96 def stream_handler(next_handler, *args, **kwargs): 97 cherrypy.response.output = o = StringIO.StringIO() 98 try: 99 response = next_handler(*args, **kwargs) 100 # Ignore the response and return our accumulated output instead. 101 return o.getvalue() 102 finally: 103 o.close() 104 cherrypy.tools.streamer = cherrypy._cptools.HandlerWrapperTool(stream_handler) 105 106 class Root: 107 def index(self): 108 return "Howdy earth!" 109 index.exposed = True 110 111 def tarfile(self): 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} 116 117 def euro(self): 118 hooks = list(cherrypy.request.hooks['before_finalize']) 119 hooks.sort() 120 assert [x.callback.__name__ for x in hooks] == ['encode', 'gzip'] 121 assert [x.priority for x in hooks] == [70, 80] 122 yield u"Hello," 123 yield u"world" 124 yield europoundUnicode 125 euro.exposed = True 126 127 # Bare hooks 128 def pipe(self): 129 return cherrypy.request.body 130 pipe.exposed = True 131 pipe._cp_config = {'hooks.before_request_body': pipe_body} 132 133 # Multiple decorators; include kwargs just for fun. 134 # Note that encode must run before gzip. 135 def decorated_euro(self, *vpath): 136 yield u"Hello," 137 yield u"world" 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) 142 143 root = Root() 144 145 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. 149 """ 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): 154 value.exposed = True 155 setattr(root, name.lower(), cls()) 156 class Test(object): 157 __metaclass__ = TestType 158 159 160 # METHOD ONE: 161 # Declare Tools in _cp_config 162 class Demo(Test): 163 164 _cp_config = {"tools.nadsat.on": True} 165 166 def index(self, id=None): 167 return "A good piece of cherry pie" 168 169 def ended(self, id): 170 return repr(tools.nadsat.ended[id]) 171 172 def err(self, id=None): 173 raise ValueError() 174 175 def errinstream(self, id=None): 176 yield "nonconfidential" 177 raise ValueError() 178 yield "confidential" 179 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): 184 return "Welcome!" 185 restricted = myauthtools.check_access()(restricted) 186 userid = restricted 187 188 def err_in_onstart(self): 189 return "success!" 190 191 def stream(self, id=None): 192 for x in xrange(100000000): 193 yield str(x) 194 stream._cp_config = {'response.stream': True} 195 196 197 cherrypy.config.update({'environment': 'test_suite'}) 198 199 conf = { 200 # METHOD THREE: 201 # Declare Tools in detached config 202 '/demo': { 203 'tools.numerify.on': True, 204 'tools.numerify.map': {"pie": "3.14159"}, 205 }, 206 '/demo/restricted': { 207 'request.show_tracebacks': False, 208 }, 209 '/demo/userid': { 210 'request.show_tracebacks': False, 211 'myauth.check_access.default': True, 212 }, 213 '/demo/errinstream': { 214 'response.stream': True, 215 }, 216 '/demo/err_in_onstart': { 217 # Because this isn't a dict, on_start_resource will error. 218 'tools.numerify.map': "pie->3.14159" 219 }, 220 # Combined tools 221 '/euro': { 222 'tools.gzip.on': True, 223 'tools.encode.on': True, 224 }, 225 # Priority specified in config 226 '/decorated_euro/subpath': { 227 'tools.gzip.priority': 10, 228 }, 229 # Handler wrappers 230 '/tarfile': {'tools.streamer.on': True} 231 } 232 app = cherrypy.tree.mount(root, config=conf) 233 app.request_class.namespaces['myauth'] = myauthtools 234 235 if sys.version_info >= (2, 5): 236 from cherrypy.test import py25 237 root.tooldecs = py25.ToolExamples() 238 239 240 # Client-side code # 241 242 from cherrypy.test import helper 243 244
245 -class ToolTests(helper.CPWebCase):
246
247 - def testHookErrors(self):
248 self.getPage("/demo/?id=1") 249 # If body is "razdrez", then on_end_request is being called too early. 250 self.assertBody("A horrorshow lomtick of cherry 3.14159") 251 # If this fails, then on_end_request isn't being called at all. 252 time.sleep(0.1) 253 self.getPage("/demo/ended/1") 254 self.assertBody("True") 255 256 valerr = '\n raise ValueError()\nValueError' 257 self.getPage("/demo/err?id=3") 258 # If body is "razdrez", then on_end_request is being called too early. 259 self.assertErrorPage(502, pattern=valerr) 260 # If this fails, then on_end_request isn't being called at all. 261 time.sleep(0.1) 262 self.getPage("/demo/ended/3") 263 self.assertBody("True") 264 265 # If body is "razdrez", then on_end_request is being called too early. 266 if (cherrypy.server.protocol_version == "HTTP/1.0" or 267 getattr(cherrypy.server, "using_apache", False)): 268 self.getPage("/demo/errinstream?id=5") 269 # Because this error is raised after the response body has 270 # started, the status should not change to an error status. 271 self.assertStatus("200 OK") 272 self.assertBody("nonconfidential") 273 else: 274 # Because this error is raised after the response body has 275 # started, and because it's chunked output, an error is raised by 276 # the HTTP client when it encounters incomplete output. 277 self.assertRaises((ValueError, IncompleteRead), self.getPage, 278 "/demo/errinstream?id=5") 279 # If this fails, then on_end_request isn't being called at all. 280 time.sleep(0.1) 281 self.getPage("/demo/ended/5") 282 self.assertBody("True") 283 284 # Test the "__call__" technique (compile-time decorator). 285 self.getPage("/demo/restricted") 286 self.assertErrorPage(401) 287 288 # Test compile-time decorator with kwargs from config. 289 self.getPage("/demo/userid") 290 self.assertBody("Welcome!")
291
292 - def testEndRequestOnDrop(self):
293 old_timeout = None 294 try: 295 httpserver = cherrypy.server.httpserver 296 old_timeout = httpserver.timeout 297 except (AttributeError, IndexError): 298 print "skipped ", 299 return 300 301 try: 302 httpserver.timeout = timeout 303 304 # Test that on_end_request is called even if the client drops. 305 self.persistent = True 306 try: 307 conn = self.HTTP_CONN 308 conn.putrequest("GET", "/demo/stream?id=9", skip_host=True) 309 conn.putheader("Host", self.HOST) 310 conn.endheaders() 311 # Skip the rest of the request and close the conn. This will 312 # cause the server's active socket to error, which *should* 313 # result in the request being aborted, and request.close being 314 # called all the way up the stack (including WSGI middleware), 315 # eventually calling our on_end_request hook. 316 finally: 317 self.persistent = False 318 time.sleep(timeout * 2) 319 # Test that the on_end_request hook was called. 320 self.getPage("/demo/ended/9") 321 self.assertBody("True") 322 finally: 323 if old_timeout is not None: 324 httpserver.timeout = old_timeout
325
326 - def testGuaranteedHooks(self):
327 # The 'critical' on_start_resource hook is 'failsafe' (guaranteed 328 # to run even if there are failures in other on_start methods). 329 # This is NOT true of the other hooks. 330 # Here, we have set up a failure in NumerifyTool.numerify_map, 331 # but our 'critical' hook should run and set the error to 502. 332 self.getPage("/demo/err_in_onstart") 333 self.assertErrorPage(502) 334 self.assertInBody("AttributeError: 'str' object has no attribute 'items'")
335
336 - def testCombinedTools(self):
337 expectedResult = (u"Hello,world" + europoundUnicode).encode('utf-8') 338 zbuf = StringIO.StringIO() 339 zfile = gzip.GzipFile(mode='wb', fileobj=zbuf, compresslevel=9) 340 zfile.write(expectedResult) 341 zfile.close() 342 343 self.getPage("/euro", headers=[("Accept-Encoding", "gzip"), 344 ("Accept-Charset", "ISO-8859-1,utf-8;q=0.7,*;q=0.7")]) 345 self.assertInBody(zbuf.getvalue()[:3]) 346 347 zbuf = StringIO.StringIO() 348 zfile = gzip.GzipFile(mode='wb', fileobj=zbuf, compresslevel=6) 349 zfile.write(expectedResult) 350 zfile.close() 351 352 self.getPage("/decorated_euro", headers=[("Accept-Encoding", "gzip")]) 353 self.assertInBody(zbuf.getvalue()[:3]) 354 355 # This should break because gzip's priority was lowered in conf. 356 # Of course, we don't want breakage in production apps, 357 # but it proves the priority was changed. 358 self.getPage("/decorated_euro/subpath", 359 headers=[("Accept-Encoding", "gzip")]) 360 self.assertErrorPage(500, pattern='UnicodeEncodeError')
361
362 - def testBareHooks(self):
363 content = "bit of a pain in me gulliver" 364 self.getPage("/pipe", 365 headers=[("Content-Length", len(content)), 366 ("Content-Type", "text/plain")], 367 method="POST", body=content) 368 self.assertBody(content)
369
370 - def testHandlerWrapperTool(self):
371 self.getPage("/tarfile") 372 self.assertBody("I am a tarfile")
373
374 - def testToolWithConfig(self):
375 if not sys.version_info >= (2, 5): 376 print "skipped (Python 2.5+ only)", 377 return 378 379 self.getPage('/tooldecs/blah') 380 self.assertHeader('Content-Type', 'application/data')
381 382 383 if __name__ == '__main__': 384 setup_server() 385 helper.testmain() 386