testing_test.py
Go to the documentation of this file.
00001 #!/usr/bin/env python
00002 
00003 from __future__ import absolute_import, division, print_function, with_statement
00004 
00005 from tornado import gen, ioloop
00006 from tornado.testing import AsyncTestCase, gen_test
00007 from tornado.test.util import unittest
00008 
00009 import contextlib
00010 import os
00011 import traceback
00012 
00013 
00014 @contextlib.contextmanager
00015 def set_environ(name, value):
00016     old_value = os.environ.get(name)
00017     os.environ[name] = value
00018 
00019     try:
00020         yield
00021     finally:
00022         if old_value is None:
00023             del os.environ[name]
00024         else:
00025             os.environ[name] = old_value
00026 
00027 
00028 class AsyncTestCaseTest(AsyncTestCase):
00029     def test_exception_in_callback(self):
00030         self.io_loop.add_callback(lambda: 1 / 0)
00031         try:
00032             self.wait()
00033             self.fail("did not get expected exception")
00034         except ZeroDivisionError:
00035             pass
00036 
00037     def test_wait_timeout(self):
00038         time = self.io_loop.time
00039 
00040         # Accept default 5-second timeout, no error
00041         self.io_loop.add_timeout(time() + 0.01, self.stop)
00042         self.wait()
00043 
00044         # Timeout passed to wait()
00045         self.io_loop.add_timeout(time() + 1, self.stop)
00046         with self.assertRaises(self.failureException):
00047             self.wait(timeout=0.01)
00048 
00049         # Timeout set with environment variable
00050         self.io_loop.add_timeout(time() + 1, self.stop)
00051         with set_environ('ASYNC_TEST_TIMEOUT', '0.01'):
00052             with self.assertRaises(self.failureException):
00053                 self.wait()
00054 
00055     def test_subsequent_wait_calls(self):
00056         """
00057         This test makes sure that a second call to wait()
00058         clears the first timeout.
00059         """
00060         self.io_loop.add_timeout(self.io_loop.time() + 0.01, self.stop)
00061         self.wait(timeout=0.02)
00062         self.io_loop.add_timeout(self.io_loop.time() + 0.03, self.stop)
00063         self.wait(timeout=0.15)
00064 
00065 
00066 class AsyncTestCaseWrapperTest(unittest.TestCase):
00067     def test_undecorated_generator(self):
00068         class Test(AsyncTestCase):
00069             def test_gen(self):
00070                 yield
00071         test = Test('test_gen')
00072         result = unittest.TestResult()
00073         test.run(result)
00074         self.assertEqual(len(result.errors), 1)
00075         self.assertIn("should be decorated", result.errors[0][1])
00076 
00077     def test_undecorated_generator_with_skip(self):
00078         class Test(AsyncTestCase):
00079             @unittest.skip("don't run this")
00080             def test_gen(self):
00081                 yield
00082         test = Test('test_gen')
00083         result = unittest.TestResult()
00084         test.run(result)
00085         self.assertEqual(len(result.errors), 0)
00086         self.assertEqual(len(result.skipped), 1)
00087 
00088     def test_other_return(self):
00089         class Test(AsyncTestCase):
00090             def test_other_return(self):
00091                 return 42
00092         test = Test('test_other_return')
00093         result = unittest.TestResult()
00094         test.run(result)
00095         self.assertEqual(len(result.errors), 1)
00096         self.assertIn("Return value from test method ignored", result.errors[0][1])
00097 
00098 
00099 class SetUpTearDownTest(unittest.TestCase):
00100     def test_set_up_tear_down(self):
00101         """
00102         This test makes sure that AsyncTestCase calls super methods for
00103         setUp and tearDown.
00104 
00105         InheritBoth is a subclass of both AsyncTestCase and
00106         SetUpTearDown, with the ordering so that the super of
00107         AsyncTestCase will be SetUpTearDown.
00108         """
00109         events = []
00110         result = unittest.TestResult()
00111 
00112         class SetUpTearDown(unittest.TestCase):
00113             def setUp(self):
00114                 events.append('setUp')
00115 
00116             def tearDown(self):
00117                 events.append('tearDown')
00118 
00119         class InheritBoth(AsyncTestCase, SetUpTearDown):
00120             def test(self):
00121                 events.append('test')
00122 
00123         InheritBoth('test').run(result)
00124         expected = ['setUp', 'test', 'tearDown']
00125         self.assertEqual(expected, events)
00126 
00127 
00128 class GenTest(AsyncTestCase):
00129     def setUp(self):
00130         super(GenTest, self).setUp()
00131         self.finished = False
00132 
00133     def tearDown(self):
00134         self.assertTrue(self.finished)
00135         super(GenTest, self).tearDown()
00136 
00137     @gen_test
00138     def test_sync(self):
00139         self.finished = True
00140 
00141     @gen_test
00142     def test_async(self):
00143         yield gen.Task(self.io_loop.add_callback)
00144         self.finished = True
00145 
00146     def test_timeout(self):
00147         # Set a short timeout and exceed it.
00148         @gen_test(timeout=0.1)
00149         def test(self):
00150             yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)
00151 
00152         # This can't use assertRaises because we need to inspect the
00153         # exc_info triple (and not just the exception object)
00154         try:
00155             test(self)
00156             self.fail("did not get expected exception")
00157         except ioloop.TimeoutError:
00158             # The stack trace should blame the add_timeout line, not just
00159             # unrelated IOLoop/testing internals.
00160             self.assertIn(
00161                 "gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
00162                 traceback.format_exc())
00163 
00164         self.finished = True
00165 
00166     def test_no_timeout(self):
00167         # A test that does not exceed its timeout should succeed.
00168         @gen_test(timeout=1)
00169         def test(self):
00170             time = self.io_loop.time
00171             yield gen.Task(self.io_loop.add_timeout, time() + 0.1)
00172 
00173         test(self)
00174         self.finished = True
00175 
00176     def test_timeout_environment_variable(self):
00177         @gen_test(timeout=0.5)
00178         def test_long_timeout(self):
00179             time = self.io_loop.time
00180             yield gen.Task(self.io_loop.add_timeout, time() + 0.25)
00181 
00182         # Uses provided timeout of 0.5 seconds, doesn't time out.
00183         with set_environ('ASYNC_TEST_TIMEOUT', '0.1'):
00184             test_long_timeout(self)
00185 
00186         self.finished = True
00187 
00188     def test_no_timeout_environment_variable(self):
00189         @gen_test(timeout=0.01)
00190         def test_short_timeout(self):
00191             time = self.io_loop.time
00192             yield gen.Task(self.io_loop.add_timeout, time() + 1)
00193 
00194         # Uses environment-variable timeout of 0.1, times out.
00195         with set_environ('ASYNC_TEST_TIMEOUT', '0.1'):
00196             with self.assertRaises(ioloop.TimeoutError):
00197                 test_short_timeout(self)
00198 
00199         self.finished = True
00200 
00201     def test_with_method_args(self):
00202         @gen_test
00203         def test_with_args(self, *args):
00204             self.assertEqual(args, ('test',))
00205             yield gen.Task(self.io_loop.add_callback)
00206 
00207         test_with_args(self, 'test')
00208         self.finished = True
00209 
00210     def test_with_method_kwargs(self):
00211         @gen_test
00212         def test_with_kwargs(self, **kwargs):
00213             self.assertDictEqual(kwargs, {'test': 'test'})
00214             yield gen.Task(self.io_loop.add_callback)
00215 
00216         test_with_kwargs(self, test='test')
00217         self.finished = True
00218 
00219 if __name__ == '__main__':
00220     unittest.main()


rosbridge_server
Author(s): Jonathan Mace
autogenerated on Thu Jun 6 2019 21:51:51