Source

CherryPy / cherrypy / test / test_tools.py

Diff from to

File cherrypy/test/test_tools.py

 
 class ToolTests(helper.CPWebCase):
     def setup_server():
-        
+
         # Put check_access in a custom toolbox with its own namespace
         myauthtools = cherrypy._cptools.Toolbox("myauth")
-        
+
         def check_access(default=False):
             if not getattr(cherrypy.request, "userid", default):
                 raise cherrypy.HTTPError(401)
         myauthtools.check_access = cherrypy.Tool('before_request_body', check_access)
-        
+
         def numerify():
             def number_it(body):
                 for chunk in body:
                         chunk = chunk.replace(k, v)
                     yield chunk
             cherrypy.response.body = number_it(cherrypy.response.body)
-        
+
         class NumTool(cherrypy.Tool):
             def _setup(self):
                 def makemap():
                     m = self._merged_args().get("map", {})
                     cherrypy.request.numerify_map = copyitems(m)
                 cherrypy.request.hooks.attach('on_start_resource', makemap)
-                
+
                 def critical():
                     cherrypy.request.error_response = cherrypy.HTTPError(502).set_response
                 critical.failsafe = True
-                
+
                 cherrypy.request.hooks.attach('on_start_resource', critical)
                 cherrypy.request.hooks.attach(self._point, self.callable)
-        
+
         tools.numerify = NumTool('before_finalize', numerify)
-        
+
         # It's not mandatory to inherit from cherrypy.Tool.
         class NadsatTool:
-            
+
             def __init__(self):
                 self.ended = {}
                 self._name = "nadsat"
-            
+
             def nadsat(self):
                 def nadsat_it_up(body):
                     for chunk in body:
                         yield chunk
                 cherrypy.response.body = nadsat_it_up(cherrypy.response.body)
             nadsat.priority = 0
-            
+
             def cleanup(self):
                 # This runs after the request has been completely written out.
                 cherrypy.response.body = [ntob("razdrez")]
                 if id:
                     self.ended[id] = True
             cleanup.failsafe = True
-            
+
             def _setup(self):
                 cherrypy.request.hooks.attach('before_finalize', self.nadsat)
                 cherrypy.request.hooks.attach('on_end_request', self.cleanup)
         tools.nadsat = NadsatTool()
-        
+
         def pipe_body():
             cherrypy.request.process_request_body = False
             clen = int(cherrypy.request.headers['Content-Length'])
             cherrypy.request.body = cherrypy.request.rfile.read(clen)
-        
+
         # Assert that we can use a callable object instead of a function.
         class Rotator(object):
             def __call__(self, scale):
                 else:
                     r.body = [chr((ord(x) + scale) % 256) for x in r.body[0]]
         cherrypy.tools.rotator = cherrypy.Tool('before_finalize', Rotator())
-        
+
         def stream_handler(next_handler, *args, **kwargs):
             cherrypy.response.output = o = BytesIO()
             try:
             finally:
                 o.close()
         cherrypy.tools.streamer = cherrypy._cptools.HandlerWrapperTool(stream_handler)
-        
+
         class Root:
             def index(self):
                 return "Howdy earth!"
             index.exposed = True
-            
+
             def tarfile(self):
                 cherrypy.response.output.write(ntob('I am '))
                 cherrypy.response.output.write(ntob('a tarfile'))
             tarfile.exposed = True
             tarfile._cp_config = {'tools.streamer.on': True}
-            
+
             def euro(self):
                 hooks = list(cherrypy.request.hooks['before_finalize'])
                 hooks.sort()
                 yield ntou("world")
                 yield europoundUnicode
             euro.exposed = True
-            
+
             # Bare hooks
             def pipe(self):
                 return cherrypy.request.body
             pipe.exposed = True
             pipe._cp_config = {'hooks.before_request_body': pipe_body}
-            
+
             # Multiple decorators; include kwargs just for fun.
             # Note that rotator must run before gzip.
             def decorated_euro(self, *vpath):
             decorated_euro.exposed = True
             decorated_euro = tools.gzip(compress_level=6)(decorated_euro)
             decorated_euro = tools.rotator(scale=3)(decorated_euro)
-        
+
         root = Root()
-        
-        
+
         class TestType(type):
             """Metaclass which automatically exposes all functions in each subclass,
             and adds an instance of the subclass as an attribute of root.
                         value.exposed = True
                 setattr(root, name.lower(), cls())
         Test = TestType('Test', (object,), {})
-        
-        
+
         # METHOD ONE:
         # Declare Tools in _cp_config
         class Demo(Test):
-            
+
             _cp_config = {"tools.nadsat.on": True}
-            
+
             def index(self, id=None):
                 return "A good piece of cherry pie"
-            
+
             def ended(self, id):
                 return repr(tools.nadsat.ended[id])
-            
+
             def err(self, id=None):
                 raise ValueError()
-            
+
             def errinstream(self, id=None):
                 yield "nonconfidential"
                 raise ValueError()
                 yield "confidential"
-            
+
             # METHOD TWO: decorator using Tool()
             # We support Python 2.3, but the @-deco syntax would look like this:
             # @tools.check_access()
                 return "Welcome!"
             restricted = myauthtools.check_access()(restricted)
             userid = restricted
-            
+
             def err_in_onstart(self):
                 return "success!"
-            
+
             def stream(self, id=None):
                 for x in xrange(100000000):
                     yield str(x)
             stream._cp_config = {'response.stream': True}
-        
-        
+
+
         conf = {
             # METHOD THREE:
             # Declare Tools in detached config
         }
         app = cherrypy.tree.mount(root, config=conf)
         app.request_class.namespaces['myauth'] = myauthtools
-        
+
         if sys.version_info >= (2, 5):
             from cherrypy.test import _test_decorators
             root.tooldecs = _test_decorators.ToolExamples()
         time.sleep(0.1)
         self.getPage("/demo/ended/1")
         self.assertBody("True")
-        
+
         valerr = '\n    raise ValueError()\nValueError'
         self.getPage("/demo/err?id=3")
         # If body is "razdrez", then on_end_request is being called too early.
         time.sleep(0.1)
         self.getPage("/demo/ended/3")
         self.assertBody("True")
-        
+
         # If body is "razdrez", then on_end_request is being called too early.
         if (cherrypy.server.protocol_version == "HTTP/1.0" or
             getattr(cherrypy.server, "using_apache", False)):
         time.sleep(0.1)
         self.getPage("/demo/ended/5")
         self.assertBody("True")
-        
+
         # Test the "__call__" technique (compile-time decorator).
         self.getPage("/demo/restricted")
         self.assertErrorPage(401)
-        
+
         # Test compile-time decorator with kwargs from config.
         self.getPage("/demo/userid")
         self.assertBody("Welcome!")
-    
+
     def testEndRequestOnDrop(self):
         old_timeout = None
         try:
             old_timeout = httpserver.timeout
         except (AttributeError, IndexError):
             return self.skip()
-        
+
         try:
             httpserver.timeout = timeout
-            
+
             # Test that on_end_request is called even if the client drops.
             self.persistent = True
             try:
         finally:
             if old_timeout is not None:
                 httpserver.timeout = old_timeout
-    
+
     def testGuaranteedHooks(self):
         # The 'critical' on_start_resource hook is 'failsafe' (guaranteed
         # to run even if there are failures in other on_start methods).
         self.getPage("/demo/err_in_onstart")
         self.assertErrorPage(502)
         self.assertInBody("AttributeError: 'str' object has no attribute 'items'")
-    
+
     def testCombinedTools(self):
         expectedResult = (ntou("Hello,world") + europoundUnicode).encode('utf-8')
         zbuf = BytesIO()
         zfile = gzip.GzipFile(mode='wb', fileobj=zbuf, compresslevel=9)
         zfile.write(expectedResult)
         zfile.close()
-        
+
         self.getPage("/euro", headers=[("Accept-Encoding", "gzip"),
                                         ("Accept-Charset", "ISO-8859-1,utf-8;q=0.7,*;q=0.7")])
         self.assertInBody(zbuf.getvalue()[:3])
-        
+
         zbuf = BytesIO()
         zfile = gzip.GzipFile(mode='wb', fileobj=zbuf, compresslevel=6)
         zfile.write(expectedResult)
         zfile.close()
-        
+
         self.getPage("/decorated_euro", headers=[("Accept-Encoding", "gzip")])
         self.assertInBody(zbuf.getvalue()[:3])
-        
+
         # This returns a different value because gzip's priority was
         # lowered in conf, allowing the rotator to run after gzip.
         # Of course, we don't want breakage in production apps,
             self.assertInBody(bytes([(x + 3) % 256 for x in zbuf.getvalue()]))
         else:
             self.assertInBody(''.join([chr((ord(x) + 3) % 256) for x in zbuf.getvalue()]))
-    
+
     def testBareHooks(self):
         content = "bit of a pain in me gulliver"
         self.getPage("/pipe",
                               ("Content-Type", "text/plain")],
                      method="POST", body=content)
         self.assertBody(content)
-    
+
     def testHandlerWrapperTool(self):
         self.getPage("/tarfile")
         self.assertBody("I am a tarfile")
-    
+
     def testToolWithConfig(self):
         if not sys.version_info >= (2, 5):
             return self.skip("skipped (Python 2.5+ only)")
-        
+
         self.getPage('/tooldecs/blah')
         self.assertHeader('Content-Type', 'application/data')
-    
+
     def testWarnToolOn(self):
         # get
         try:
-            numon = cherrypy.tools.numerify.on
+            cherrypy.tools.numerify.on
         except AttributeError:
             pass
         else:
             raise AssertionError("Tool.on did not error as it should have.")
-        
+
         # set
         try:
             cherrypy.tools.numerify.on = True
             pass
         else:
             raise AssertionError("Tool.on did not error as it should have.")
-