codekingpro/portable-devtools

1from tornado import gen, ioloop
2from tornado.httpserver import HTTPServer
3from tornado.locks import Event
4from tornado.testing import AsyncHTTPTestCase, AsyncTestCase, bind_unused_port, gen_test
5from tornado.web import Application
6import asyncio
7import contextlib
8import gc
9import os
10import platform
11import sys
12import traceback
13import unittest
14import warnings
15
16
17@contextlib.contextmanager
18def set_environ(name, value):
19    old_value = os.environ.get(name)
20    os.environ[name] = value
21
22    try:
23        yield
24    finally:
25        if old_value is None:
26            del os.environ[name]
27        else:
28            os.environ[name] = old_value
29
30
31class AsyncTestCaseTest(AsyncTestCase):
32    def test_wait_timeout(self):
33        time = self.io_loop.time
34
35        # Accept default 5-second timeout, no error
36        self.io_loop.add_timeout(time() + 0.01, self.stop)
37        self.wait()
38
39        # Timeout passed to wait()
40        self.io_loop.add_timeout(time() + 1, self.stop)
41        with self.assertRaises(self.failureException):
42            self.wait(timeout=0.01)
43
44        # Timeout set with environment variable
45        self.io_loop.add_timeout(time() + 1, self.stop)
46        with set_environ("ASYNC_TEST_TIMEOUT", "0.01"):
47            with self.assertRaises(self.failureException):
48                self.wait()
49
50    def test_subsequent_wait_calls(self):
51        """
52        This test makes sure that a second call to wait()
53        clears the first timeout.
54        """
55        # The first wait ends with time left on the clock
56        self.io_loop.add_timeout(self.io_loop.time() + 0.00, self.stop)
57        self.wait(timeout=0.1)
58        # The second wait has enough time for itself but would fail if the
59        # first wait's deadline were still in effect.
60        self.io_loop.add_timeout(self.io_loop.time() + 0.2, self.stop)
61        self.wait(timeout=0.4)
62
63
64class LeakTest(AsyncTestCase):
65    def tearDown(self):
66        super().tearDown()
67        # Trigger a gc to make warnings more deterministic.
68        gc.collect()
69
70    def test_leaked_coroutine(self):
71        # This test verifies that "leaked" coroutines are shut down
72        # without triggering warnings like "task was destroyed but it
73        # is pending". If this test were to fail, it would fail
74        # because runtests.py detected unexpected output to stderr.
75        event = Event()
76
77        async def callback():
78            try:
79                await event.wait()
80            except asyncio.CancelledError:
81                pass
82
83        self.io_loop.add_callback(callback)
84        self.io_loop.add_callback(self.stop)
85        self.wait()
86
87
88class AsyncHTTPTestCaseTest(AsyncHTTPTestCase):
89    def setUp(self):
90        super().setUp()
91        # Bind a second port.
92        sock, port = bind_unused_port()
93        app = Application()
94        server = HTTPServer(app, **self.get_httpserver_options())
95        server.add_socket(sock)
96        self.second_port = port
97        self.second_server = server
98
99    def get_app(self):
100        return Application()
101
102    def test_fetch_segment(self):
103        path = "/path"
104        response = self.fetch(path)
105        self.assertEqual(response.request.url, self.get_url(path))
106
107    def test_fetch_full_http_url(self):
108        # Ensure that self.fetch() recognizes absolute urls and does
109        # not transform them into references to our main test server.
110        path = "http://127.0.0.1:%d/path" % self.second_port
111
112        response = self.fetch(path)
113        self.assertEqual(response.request.url, path)
114
115    def tearDown(self):
116        self.second_server.stop()
117        super().tearDown()
118
119
120class AsyncTestCaseReturnAssertionsTest(unittest.TestCase):
121    # These tests verify that tests that return non-None values (without being decorated with
122    # @gen_test) raise errors instead of incorrectly succeeding. These tests should be removed or
123    # updated when the _callTestMethod method is removed from AsyncTestCase (the same checks will
124    # still happen, but they'll be performed in the stdlib as DeprecationWarnings)
125    def test_undecorated_generator(self):
126        class Test(AsyncTestCase):
127            def test_gen(self):
128                yield
129
130        test = Test("test_gen")
131        result = unittest.TestResult()
132        test.run(result)
133        self.assertEqual(len(result.errors), 1)
134        self.assertIn("should be decorated", result.errors[0][1])
135
136    @unittest.skipIf(
137        platform.python_implementation() == "PyPy",
138        "pypy destructor warnings cannot be silenced",
139    )
140    @unittest.skipIf(
141        # This check actually exists in 3.11 but it changed in 3.12 in a way that breaks
142        # this test.
143        sys.version_info >= (3, 12),
144        "py312 has its own check for test case returns",
145    )
146    def test_undecorated_coroutine(self):
147        class Test(AsyncTestCase):
148            async def test_coro(self):
149                pass
150
151        test = Test("test_coro")
152        result = unittest.TestResult()
153
154        # Silence "RuntimeWarning: coroutine 'test_coro' was never awaited".
155        with warnings.catch_warnings():
156            warnings.simplefilter("ignore")
157            test.run(result)
158
159        self.assertEqual(len(result.errors), 1)
160        self.assertIn("should be decorated", result.errors[0][1])
161
162    def test_undecorated_generator_with_skip(self):
163        class Test(AsyncTestCase):
164            @unittest.skip("don't run this")
165            def test_gen(self):
166                yield
167
168        test = Test("test_gen")
169        result = unittest.TestResult()
170        test.run(result)
171        self.assertEqual(len(result.errors), 0)
172        self.assertEqual(len(result.skipped), 1)
173
174    def test_other_return(self):
175        class Test(AsyncTestCase):
176            def test_other_return(self):
177                return 42
178
179        test = Test("test_other_return")
180        result = unittest.TestResult()
181        test.run(result)
182        self.assertEqual(len(result.errors), 1)
183        self.assertIn("Return value from test method ignored", result.errors[0][1])
184
185
186class SetUpTearDownTest(unittest.TestCase):
187    def test_set_up_tear_down(self):
188        """
189        This test makes sure that AsyncTestCase calls super methods for
190        setUp and tearDown.
191
192        InheritBoth is a subclass of both AsyncTestCase and
193        SetUpTearDown, with the ordering so that the super of
194        AsyncTestCase will be SetUpTearDown.
195        """
196        events = []
197        result = unittest.TestResult()
198
199        class SetUpTearDown(unittest.TestCase):
200            def setUp(self):
201                events.append("setUp")
202
203            def tearDown(self):
204                events.append("tearDown")
205
206        class InheritBoth(AsyncTestCase, SetUpTearDown):
207            def test(self):
208                events.append("test")
209
210        InheritBoth("test").run(result)
211        expected = ["setUp", "test", "tearDown"]
212        self.assertEqual(expected, events)
213
214
215class AsyncHTTPTestCaseSetUpTearDownTest(unittest.TestCase):
216    def test_tear_down_releases_app_and_http_server(self):
217        result = unittest.TestResult()
218
219        class SetUpTearDown(AsyncHTTPTestCase):
220            def get_app(self):
221                return Application()
222
223            def test(self):
224                self.assertTrue(hasattr(self, "_app"))
225                self.assertTrue(hasattr(self, "http_server"))
226
227        test = SetUpTearDown("test")
228        test.run(result)
229        self.assertFalse(hasattr(test, "_app"))
230        self.assertFalse(hasattr(test, "http_server"))
231
232
233class GenTest(AsyncTestCase):
234    def setUp(self):
235        super().setUp()
236        self.finished = False
237
238    def tearDown(self):
239        self.assertTrue(self.finished)
240        super().tearDown()
241
242    @gen_test
243    def test_sync(self):
244        self.finished = True
245
246    @gen_test
247    def test_async(self):
248        yield gen.moment
249        self.finished = True
250
251    def test_timeout(self):
252        # Set a short timeout and exceed it.
253        @gen_test(timeout=0.1)
254        def test(self):
255            yield gen.sleep(1)
256
257        # This can't use assertRaises because we need to inspect the
258        # exc_info triple (and not just the exception object)
259        try:
260            test(self)
261            self.fail("did not get expected exception")
262        except ioloop.TimeoutError:
263            # The stack trace should blame the add_timeout line, not just
264            # unrelated IOLoop/testing internals.
265            self.assertIn("gen.sleep(1)", traceback.format_exc())
266
267        self.finished = True
268
269    def test_no_timeout(self):
270        # A test that does not exceed its timeout should succeed.
271        @gen_test(timeout=1)
272        def test(self):
273            yield gen.sleep(0.1)
274
275        test(self)
276        self.finished = True
277
278    def test_timeout_environment_variable(self):
279        @gen_test(timeout=0.5)
280        def test_long_timeout(self):
281            yield gen.sleep(0.25)
282
283        # Uses provided timeout of 0.5 seconds, doesn't time out.
284        with set_environ("ASYNC_TEST_TIMEOUT", "0.1"):
285            test_long_timeout(self)
286
287        self.finished = True
288
289    def test_no_timeout_environment_variable(self):
290        @gen_test(timeout=0.01)
291        def test_short_timeout(self):
292            yield gen.sleep(1)
293
294        # Uses environment-variable timeout of 0.1, times out.
295        with set_environ("ASYNC_TEST_TIMEOUT", "0.1"):
296            with self.assertRaises(ioloop.TimeoutError):
297                test_short_timeout(self)
298
299        self.finished = True
300
301    def test_with_method_args(self):
302        @gen_test
303        def test_with_args(self, *args):
304            self.assertEqual(args, ("test",))
305            yield gen.moment
306
307        test_with_args(self, "test")
308        self.finished = True
309
310    def test_with_method_kwargs(self):
311        @gen_test
312        def test_with_kwargs(self, **kwargs):
313            self.assertDictEqual(kwargs, {"test": "test"})
314            yield gen.moment
315
316        test_with_kwargs(self, test="test")
317        self.finished = True
318
319    def test_native_coroutine(self):
320        @gen_test
321        async def test(self):
322            self.finished = True
323
324        test(self)
325
326    def test_native_coroutine_timeout(self):
327        # Set a short timeout and exceed it.
328        @gen_test(timeout=0.1)
329        async def test(self):
330            await gen.sleep(1)
331
332        try:
333            test(self)
334            self.fail("did not get expected exception")
335        except ioloop.TimeoutError:
336            self.finished = True
337
338
339if __name__ == "__main__":
340    unittest.main()
341