monitor/cherrypy/test/test_tools.py

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