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