stack_context_test.py
Go to the documentation of this file.
00001 #!/usr/bin/env python
00002 from __future__ import absolute_import, division, with_statement
00003 
00004 from tornado.stack_context import StackContext, wrap
00005 from tornado.testing import AsyncHTTPTestCase, AsyncTestCase, LogTrapTestCase
00006 from tornado.util import b
00007 from tornado.web import asynchronous, Application, RequestHandler
00008 import contextlib
00009 import functools
00010 import logging
00011 import unittest
00012 
00013 
00014 class TestRequestHandler(RequestHandler):
00015     def __init__(self, app, request, io_loop):
00016         super(TestRequestHandler, self).__init__(app, request)
00017         self.io_loop = io_loop
00018 
00019     @asynchronous
00020     def get(self):
00021         logging.info('in get()')
00022         # call self.part2 without a self.async_callback wrapper.  Its
00023         # exception should still get thrown
00024         self.io_loop.add_callback(self.part2)
00025 
00026     def part2(self):
00027         logging.info('in part2()')
00028         # Go through a third layer to make sure that contexts once restored
00029         # are again passed on to future callbacks
00030         self.io_loop.add_callback(self.part3)
00031 
00032     def part3(self):
00033         logging.info('in part3()')
00034         raise Exception('test exception')
00035 
00036     def get_error_html(self, status_code, **kwargs):
00037         if 'exception' in kwargs and str(kwargs['exception']) == 'test exception':
00038             return 'got expected exception'
00039         else:
00040             return 'unexpected failure'
00041 
00042 
00043 class HTTPStackContextTest(AsyncHTTPTestCase, LogTrapTestCase):
00044     def get_app(self):
00045         return Application([('/', TestRequestHandler,
00046                              dict(io_loop=self.io_loop))])
00047 
00048     def test_stack_context(self):
00049         self.http_client.fetch(self.get_url('/'), self.handle_response)
00050         self.wait()
00051         self.assertEqual(self.response.code, 500)
00052         self.assertTrue(b('got expected exception') in self.response.body)
00053 
00054     def handle_response(self, response):
00055         self.response = response
00056         self.stop()
00057 
00058 
00059 class StackContextTest(AsyncTestCase, LogTrapTestCase):
00060     def setUp(self):
00061         super(StackContextTest, self).setUp()
00062         self.active_contexts = []
00063 
00064     @contextlib.contextmanager
00065     def context(self, name):
00066         self.active_contexts.append(name)
00067         yield
00068         self.assertEqual(self.active_contexts.pop(), name)
00069 
00070     # Simulates the effect of an asynchronous library that uses its own
00071     # StackContext internally and then returns control to the application.
00072     def test_exit_library_context(self):
00073         def library_function(callback):
00074             # capture the caller's context before introducing our own
00075             callback = wrap(callback)
00076             with StackContext(functools.partial(self.context, 'library')):
00077                 self.io_loop.add_callback(
00078                   functools.partial(library_inner_callback, callback))
00079 
00080         def library_inner_callback(callback):
00081             self.assertEqual(self.active_contexts[-2:],
00082                              ['application', 'library'])
00083             callback()
00084 
00085         def final_callback():
00086             # implementation detail:  the full context stack at this point
00087             # is ['application', 'library', 'application'].  The 'library'
00088             # context was not removed, but is no longer innermost so
00089             # the application context takes precedence.
00090             self.assertEqual(self.active_contexts[-1], 'application')
00091             self.stop()
00092         with StackContext(functools.partial(self.context, 'application')):
00093             library_function(final_callback)
00094         self.wait()
00095 
00096     def test_deactivate(self):
00097         deactivate_callbacks = []
00098 
00099         def f1():
00100             with StackContext(functools.partial(self.context, 'c1')) as c1:
00101                 deactivate_callbacks.append(c1)
00102                 self.io_loop.add_callback(f2)
00103 
00104         def f2():
00105             with StackContext(functools.partial(self.context, 'c2')) as c2:
00106                 deactivate_callbacks.append(c2)
00107                 self.io_loop.add_callback(f3)
00108 
00109         def f3():
00110             with StackContext(functools.partial(self.context, 'c3')) as c3:
00111                 deactivate_callbacks.append(c3)
00112                 self.io_loop.add_callback(f4)
00113 
00114         def f4():
00115             self.assertEqual(self.active_contexts, ['c1', 'c2', 'c3'])
00116             deactivate_callbacks[1]()
00117             # deactivating a context doesn't remove it immediately,
00118             # but it will be missing from the next iteration
00119             self.assertEqual(self.active_contexts, ['c1', 'c2', 'c3'])
00120             self.io_loop.add_callback(f5)
00121 
00122         def f5():
00123             self.assertEqual(self.active_contexts, ['c1', 'c3'])
00124             self.stop()
00125         self.io_loop.add_callback(f1)
00126         self.wait()
00127 
00128 if __name__ == '__main__':
00129     unittest.main()


rosbridge_server
Author(s): Jonathan Mace
autogenerated on Thu Jan 2 2014 11:53:55