diff options
Diffstat (limited to 'cherrypy/test/test_tools.py')
| -rwxr-xr-x | cherrypy/test/test_tools.py | 393 |
1 files changed, 393 insertions, 0 deletions
diff --git a/cherrypy/test/test_tools.py b/cherrypy/test/test_tools.py new file mode 100755 index 0000000..bc8579f --- /dev/null +++ b/cherrypy/test/test_tools.py @@ -0,0 +1,393 @@ +"""Test the various means of instantiating and invoking tools.""" + +import gzip +import sys +from cherrypy._cpcompat import BytesIO, copyitems, itervalues, IncompleteRead, ntob, ntou, xrange +import time +timeout = 0.2 +import types + +import cherrypy +from cherrypy import tools + + +europoundUnicode = ntou('\x80\xa3') + + +# Client-side code # + +from cherrypy.test import helper + + +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: + for k, v in cherrypy.request.numerify_map: + 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: + chunk = chunk.replace(ntob("good"), ntob("horrorshow")) + chunk = chunk.replace(ntob("piece"), ntob("lomtick")) + 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")] + id = cherrypy.request.params.get("id") + 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): + r = cherrypy.response + r.collapse_body() + 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: + response = next_handler(*args, **kwargs) + # Ignore the response and return our accumulated output instead. + return o.getvalue() + 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() + cbnames = [x.callback.__name__ for x in hooks] + assert cbnames == ['gzip'], cbnames + priorities = [x.priority for x in hooks] + assert priorities == [80], priorities + yield ntou("Hello,") + 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): + yield ntou("Hello,") + yield ntou("world") + yield europoundUnicode + 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. + """ + def __init__(cls, name, bases, dct): + type.__init__(cls, name, bases, dct) + for value in itervalues(dct): + if isinstance(value, types.FunctionType): + value.exposed = True + setattr(root, name.lower(), cls()) + class Test(object): + __metaclass__ = TestType + + + # 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() + def restricted(self): + 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 + '/demo': { + 'tools.numerify.on': True, + 'tools.numerify.map': {ntob("pie"): ntob("3.14159")}, + }, + '/demo/restricted': { + 'request.show_tracebacks': False, + }, + '/demo/userid': { + 'request.show_tracebacks': False, + 'myauth.check_access.default': True, + }, + '/demo/errinstream': { + 'response.stream': True, + }, + '/demo/err_in_onstart': { + # Because this isn't a dict, on_start_resource will error. + 'tools.numerify.map': "pie->3.14159" + }, + # Combined tools + '/euro': { + 'tools.gzip.on': True, + 'tools.encode.on': True, + }, + # Priority specified in config + '/decorated_euro/subpath': { + 'tools.gzip.priority': 10, + }, + # Handler wrappers + '/tarfile': {'tools.streamer.on': True} + } + 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() + setup_server = staticmethod(setup_server) + + def testHookErrors(self): + self.getPage("/demo/?id=1") + # If body is "razdrez", then on_end_request is being called too early. + self.assertBody("A horrorshow lomtick of cherry 3.14159") + # If this fails, then on_end_request isn't being called at all. + 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. + self.assertErrorPage(502, pattern=valerr) + # If this fails, then on_end_request isn't being called at all. + 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)): + self.getPage("/demo/errinstream?id=5") + # Because this error is raised after the response body has + # started, the status should not change to an error status. + self.assertStatus("200 OK") + self.assertBody("nonconfidential") + else: + # Because this error is raised after the response body has + # started, and because it's chunked output, an error is raised by + # the HTTP client when it encounters incomplete output. + self.assertRaises((ValueError, IncompleteRead), self.getPage, + "/demo/errinstream?id=5") + # If this fails, then on_end_request isn't being called at all. + 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: + httpserver = cherrypy.server.httpserver + 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: + conn = self.HTTP_CONN + conn.putrequest("GET", "/demo/stream?id=9", skip_host=True) + conn.putheader("Host", self.HOST) + conn.endheaders() + # Skip the rest of the request and close the conn. This will + # cause the server's active socket to error, which *should* + # result in the request being aborted, and request.close being + # called all the way up the stack (including WSGI middleware), + # eventually calling our on_end_request hook. + finally: + self.persistent = False + time.sleep(timeout * 2) + # Test that the on_end_request hook was called. + self.getPage("/demo/ended/9") + self.assertBody("True") + 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). + # This is NOT true of the other hooks. + # Here, we have set up a failure in NumerifyTool.numerify_map, + # but our 'critical' hook should run and set the error to 502. + 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, + # but it proves the priority was changed. + self.getPage("/decorated_euro/subpath", + headers=[("Accept-Encoding", "gzip")]) + 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", + headers=[("Content-Length", str(len(content))), + ("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 + except AttributeError: + pass + else: + raise AssertionError("Tool.on did not error as it should have.") + + # set + try: + cherrypy.tools.numerify.on = True + except AttributeError: + pass + else: + raise AssertionError("Tool.on did not error as it should have.") + |