codekingpro/portable-devtools

1# Licensed under the Apache License, Version 2.0 (the "License"); you may
2# not use this file except in compliance with the License. You may obtain
3# a copy of the License at
4#
5#     http://www.apache.org/licenses/LICENSE-2.0
6#
7# Unless required by applicable law or agreed to in writing, software
8# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
9# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
10# License for the specific language governing permissions and limitations
11# under the License.
12
13import asyncio
14import contextvars
15import threading
16import time
17import unittest
18import warnings
19
20from concurrent.futures import ThreadPoolExecutor
21import tornado.platform.asyncio
22from tornado import gen
23from tornado.ioloop import IOLoop
24from tornado.platform.asyncio import (
25    AsyncIOLoop,
26    to_asyncio_future,
27    AddThreadSelectorEventLoop,
28)
29from tornado.testing import (
30    AsyncTestCase,
31    gen_test,
32    setup_with_context_manager,
33    AsyncHTTPTestCase,
34)
35from tornado.test.util import ignore_deprecation
36from tornado.web import Application, RequestHandler
37
38
39class AsyncIOLoopTest(AsyncTestCase):
40    @property
41    def asyncio_loop(self):
42        return self.io_loop.asyncio_loop  # type: ignore
43
44    def test_asyncio_callback(self):
45        # Basic test that the asyncio loop is set up correctly.
46        async def add_callback():
47            asyncio.get_event_loop().call_soon(self.stop)
48
49        self.asyncio_loop.run_until_complete(add_callback())
50        self.wait()
51
52    @gen_test
53    def test_asyncio_future(self):
54        # Test that we can yield an asyncio future from a tornado coroutine.
55        # Without 'yield from', we must wrap coroutines in ensure_future.
56        x = yield asyncio.ensure_future(
57            asyncio.get_event_loop().run_in_executor(None, lambda: 42)
58        )
59        self.assertEqual(x, 42)
60
61    @gen_test
62    def test_asyncio_yield_from(self):
63        @gen.coroutine
64        def f():
65            event_loop = asyncio.get_event_loop()
66            x = yield from event_loop.run_in_executor(None, lambda: 42)
67            return x
68
69        result = yield f()
70        self.assertEqual(result, 42)
71
72    def test_asyncio_adapter(self):
73        # This test demonstrates that when using the asyncio coroutine
74        # runner (i.e. run_until_complete), the to_asyncio_future
75        # adapter is needed. No adapter is needed in the other direction,
76        # as demonstrated by other tests in the package.
77        @gen.coroutine
78        def tornado_coroutine():
79            yield gen.moment
80            raise gen.Return(42)
81
82        async def native_coroutine_without_adapter():
83            return await tornado_coroutine()
84
85        async def native_coroutine_with_adapter():
86            return await to_asyncio_future(tornado_coroutine())
87
88        # Use the adapter, but two degrees from the tornado coroutine.
89        async def native_coroutine_with_adapter2():
90            return await to_asyncio_future(native_coroutine_without_adapter())
91
92        # Tornado supports native coroutines both with and without adapters
93        self.assertEqual(self.io_loop.run_sync(native_coroutine_without_adapter), 42)
94        self.assertEqual(self.io_loop.run_sync(native_coroutine_with_adapter), 42)
95        self.assertEqual(self.io_loop.run_sync(native_coroutine_with_adapter2), 42)
96
97        # Asyncio only supports coroutines that yield asyncio-compatible
98        # Futures (which our Future is since 5.0).
99        self.assertEqual(
100            self.asyncio_loop.run_until_complete(native_coroutine_without_adapter()),
101            42,
102        )
103        self.assertEqual(
104            self.asyncio_loop.run_until_complete(native_coroutine_with_adapter()),
105            42,
106        )
107        self.assertEqual(
108            self.asyncio_loop.run_until_complete(native_coroutine_with_adapter2()),
109            42,
110        )
111
112    def test_add_thread_close_idempotent(self):
113        loop = AddThreadSelectorEventLoop(asyncio.get_event_loop())  # type: ignore
114        loop.close()
115        loop.close()
116
117
118class LeakTest(unittest.TestCase):
119    def setUp(self):
120        # Trigger a cleanup of the mapping so we start with a clean slate.
121        AsyncIOLoop(make_current=False).close()
122
123    def tearDown(self):
124        try:
125            loop = asyncio.get_event_loop_policy().get_event_loop()
126        except Exception:
127            # We may not have a current event loop at this point.
128            pass
129        else:
130            loop.close()
131
132    def test_ioloop_close_leak(self):
133        orig_count = len(IOLoop._ioloop_for_asyncio)
134        for i in range(10):
135            # Create and close an AsyncIOLoop using Tornado interfaces.
136            with warnings.catch_warnings():
137                warnings.simplefilter("ignore", DeprecationWarning)
138                loop = AsyncIOLoop()
139                loop.close()
140        new_count = len(IOLoop._ioloop_for_asyncio) - orig_count
141        self.assertEqual(new_count, 0)
142
143    def test_asyncio_close_leak(self):
144        orig_count = len(IOLoop._ioloop_for_asyncio)
145        for i in range(10):
146            # Create and close an AsyncIOMainLoop using asyncio interfaces.
147            loop = asyncio.new_event_loop()
148            loop.call_soon(IOLoop.current)
149            loop.call_soon(loop.stop)
150            loop.run_forever()
151            loop.close()
152        new_count = len(IOLoop._ioloop_for_asyncio) - orig_count
153        # Because the cleanup is run on new loop creation, we have one
154        # dangling entry in the map (but only one).
155        self.assertEqual(new_count, 1)
156
157
158class SelectorThreadLeakTest(unittest.TestCase):
159    # These tests are only relevant on windows, but they should pass anywhere.
160    def setUp(self):
161        # As a precaution, ensure that we've run an event loop at least once
162        # so if it spins up any singleton threads they're already there.
163        asyncio.run(self.dummy_tornado_coroutine())
164        self.orig_thread_count = threading.active_count()
165
166    def assert_no_thread_leak(self):
167        # For some reason we see transient failures here, but I haven't been able
168        # to catch it to identify which thread is causing it. Whatever thread it
169        # is, it appears to quickly clean up on its own, so just retry a few times.
170        # At least some of the time the errant thread was running at the time we
171        # captured self.orig_thread_count, so use inequalities.
172        deadline = time.time() + 1
173        while time.time() < deadline:
174            threads = list(threading.enumerate())
175            if len(threads) <= self.orig_thread_count:
176                break
177            time.sleep(0.1)
178        self.assertLessEqual(len(threads), self.orig_thread_count, threads)
179
180    async def dummy_tornado_coroutine(self):
181        # Just access the IOLoop to initialize the selector thread.
182        IOLoop.current()
183
184    def test_asyncio_run(self):
185        for i in range(10):
186            # asyncio.run calls shutdown_asyncgens for us.
187            asyncio.run(self.dummy_tornado_coroutine())
188        self.assert_no_thread_leak()
189
190    def test_asyncio_manual(self):
191        for i in range(10):
192            loop = asyncio.new_event_loop()
193            loop.run_until_complete(self.dummy_tornado_coroutine())
194            # Without this step, we'd leak the thread.
195            loop.run_until_complete(loop.shutdown_asyncgens())
196            loop.close()
197        self.assert_no_thread_leak()
198
199    def test_tornado(self):
200        for i in range(10):
201            # The IOLoop interfaces are aware of the selector thread and
202            # (synchronously) shut it down.
203            loop = IOLoop(make_current=False)
204            loop.run_sync(self.dummy_tornado_coroutine)
205            loop.close()
206        self.assert_no_thread_leak()
207
208
209class AnyThreadEventLoopPolicyTest(unittest.TestCase):
210    def setUp(self):
211        setup_with_context_manager(self, ignore_deprecation())
212        # Referencing the event loop policy attributes raises deprecation warnings,
213        # so instead of importing this at the top of the file we capture it here.
214        self.AnyThreadEventLoopPolicy = (
215            tornado.platform.asyncio.AnyThreadEventLoopPolicy
216        )
217        self.orig_policy = asyncio.get_event_loop_policy()
218        self.executor = ThreadPoolExecutor(1)
219
220    def tearDown(self):
221        asyncio.set_event_loop_policy(self.orig_policy)
222        self.executor.shutdown()
223
224    def get_event_loop_on_thread(self):
225        def get_and_close_event_loop():
226            """Get the event loop. Close it if one is returned.
227
228            Returns the (closed) event loop. This is a silly thing
229            to do and leaves the thread in a broken state, but it's
230            enough for this test. Closing the loop avoids resource
231            leak warnings.
232            """
233            loop = asyncio.get_event_loop()
234            loop.close()
235            return loop
236
237        future = self.executor.submit(get_and_close_event_loop)
238        return future.result()
239
240    def test_asyncio_accessor(self):
241        with warnings.catch_warnings():
242            warnings.simplefilter("ignore", DeprecationWarning)
243            # With the default policy, non-main threads don't get an event
244            # loop.
245            self.assertRaises(
246                RuntimeError, self.executor.submit(asyncio.get_event_loop).result
247            )
248            # Set the policy and we can get a loop.
249            asyncio.set_event_loop_policy(self.AnyThreadEventLoopPolicy())
250            self.assertIsInstance(
251                self.executor.submit(asyncio.get_event_loop).result(),
252                asyncio.AbstractEventLoop,
253            )
254            # Clean up to silence leak warnings. Always use asyncio since
255            # IOLoop doesn't (currently) close the underlying loop.
256            self.executor.submit(lambda: asyncio.get_event_loop().close()).result()  # type: ignore
257
258    def test_tornado_accessor(self):
259        # Tornado's IOLoop.current() API can create a loop for any thread,
260        # regardless of this event loop policy.
261        with warnings.catch_warnings():
262            warnings.simplefilter("ignore", DeprecationWarning)
263            self.assertIsInstance(self.executor.submit(IOLoop.current).result(), IOLoop)
264            # Clean up to silence leak warnings. Always use asyncio since
265            # IOLoop doesn't (currently) close the underlying loop.
266            self.executor.submit(lambda: asyncio.get_event_loop().close()).result()  # type: ignore
267
268            asyncio.set_event_loop_policy(self.AnyThreadEventLoopPolicy())
269            self.assertIsInstance(self.executor.submit(IOLoop.current).result(), IOLoop)
270            self.executor.submit(lambda: asyncio.get_event_loop().close()).result()  # type: ignore
271
272
273class SelectorThreadContextvarsTest(AsyncHTTPTestCase):
274    ctx_value = "foo"
275    test_endpoint = "/"
276    tornado_test_ctx = contextvars.ContextVar("tornado_test_ctx", default="default")
277    tornado_test_ctx.set(ctx_value)
278
279    def get_app(self) -> Application:
280        tornado_test_ctx = self.tornado_test_ctx
281
282        class Handler(RequestHandler):
283            async def get(self):
284                # On the Windows platform,
285                # when a asyncio.events.Handle is created
286                # in the SelectorThread without providing a context,
287                # it will copy the current thread's context,
288                # which can lead to the loss of the main thread's context
289                # when executing the handle.
290                # Therefore, it is necessary to
291                # save a copy of the main thread's context in the SelectorThread
292                # for creating the handle.
293                self.write(tornado_test_ctx.get())
294
295        return Application([(self.test_endpoint, Handler)])
296
297    def test_context_vars(self):
298        self.assertEqual(self.ctx_value, self.fetch(self.test_endpoint).body.decode())
299