{
    "file_diffs": [
        {
            "old_file_content": "# Licensed under the Apache License, Version 2.0 (the \"License\"); you may\n# not use this file except in compliance with the License. You may obtain\n# a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT\n# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n# License for the specific language governing permissions and limitations\n# under the License.\n\nimport asyncio\nimport unittest\nimport warnings\n\nfrom concurrent.futures import ThreadPoolExecutor\nfrom tornado import gen\nfrom tornado.ioloop import IOLoop\nfrom tornado.platform.asyncio import (\n    AsyncIOLoop,\n    to_asyncio_future,\n    AnyThreadEventLoopPolicy,\n)\nfrom tornado.testing import AsyncTestCase, gen_test\n\n\nclass AsyncIOLoopTest(AsyncTestCase):\n    def get_new_ioloop(self):\n        io_loop = AsyncIOLoop(make_current=False)\n        return io_loop\n\n    def test_asyncio_callback(self):\n        # Basic test that the asyncio loop is set up correctly.\n        async def add_callback():\n            asyncio.get_event_loop().call_soon(self.stop)\n\n        self.asyncio_loop.run_until_complete(add_callback())\n        self.wait()\n\n    @gen_test\n    def test_asyncio_future(self):\n        # Test that we can yield an asyncio future from a tornado coroutine.\n        # Without 'yield from', we must wrap coroutines in ensure_future,\n        # which was introduced during Python 3.4, deprecating the prior \"async\".\n        if hasattr(asyncio, \"ensure_future\"):\n            ensure_future = asyncio.ensure_future\n        else:\n            # async is a reserved word in Python 3.7\n            ensure_future = getattr(asyncio, \"async\")\n\n        x = yield ensure_future(\n            asyncio.get_event_loop().run_in_executor(None, lambda: 42)\n        )\n        self.assertEqual(x, 42)\n\n    @gen_test\n    def test_asyncio_yield_from(self):\n        @gen.coroutine\n        def f():\n            event_loop = asyncio.get_event_loop()\n            x = yield from event_loop.run_in_executor(None, lambda: 42)\n            return x\n\n        result = yield f()\n        self.assertEqual(result, 42)\n\n    def test_asyncio_adapter(self):\n        # This test demonstrates that when using the asyncio coroutine\n        # runner (i.e. run_until_complete), the to_asyncio_future\n        # adapter is needed. No adapter is needed in the other direction,\n        # as demonstrated by other tests in the package.\n        @gen.coroutine\n        def tornado_coroutine():\n            yield gen.moment\n            raise gen.Return(42)\n\n        async def native_coroutine_without_adapter():\n            return await tornado_coroutine()\n\n        async def native_coroutine_with_adapter():\n            return await to_asyncio_future(tornado_coroutine())\n\n        # Use the adapter, but two degrees from the tornado coroutine.\n        async def native_coroutine_with_adapter2():\n            return await to_asyncio_future(native_coroutine_without_adapter())\n\n        # Tornado supports native coroutines both with and without adapters\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_without_adapter), 42)\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_with_adapter), 42)\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_with_adapter2), 42)\n\n        # Asyncio only supports coroutines that yield asyncio-compatible\n        # Futures (which our Future is since 5.0).\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_without_adapter()),\n            42,\n        )\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_with_adapter()),\n            42,\n        )\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_with_adapter2()),\n            42,\n        )\n        # I'm not entirely sure why this manual cleanup is necessary but without\n        # it we have at-a-distance failures in ioloop_test.TestIOLoopCurrent.\n        asyncio.set_event_loop(None)\n\n\nclass LeakTest(unittest.TestCase):\n    def setUp(self):\n        # Trigger a cleanup of the mapping so we start with a clean slate.\n        AsyncIOLoop(make_current=False).close()\n        # If we don't clean up after ourselves other tests may fail on\n        # py34.\n        self.orig_policy = asyncio.get_event_loop_policy()\n        asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())\n\n    def tearDown(self):\n        try:\n            loop = asyncio.get_event_loop_policy().get_event_loop()\n        except Exception:\n            # We may not have a current event loop at this point.\n            pass\n        else:\n            loop.close()\n        asyncio.set_event_loop_policy(self.orig_policy)\n\n    def test_ioloop_close_leak(self):\n        orig_count = len(IOLoop._ioloop_for_asyncio)\n        for i in range(10):\n            # Create and close an AsyncIOLoop using Tornado interfaces.\n            with warnings.catch_warnings():\n                warnings.simplefilter(\"ignore\", DeprecationWarning)\n                loop = AsyncIOLoop()\n                loop.close()\n        new_count = len(IOLoop._ioloop_for_asyncio) - orig_count\n        self.assertEqual(new_count, 0)\n\n    def test_asyncio_close_leak(self):\n        orig_count = len(IOLoop._ioloop_for_asyncio)\n        for i in range(10):\n            # Create and close an AsyncIOMainLoop using asyncio interfaces.\n            loop = asyncio.new_event_loop()\n            loop.call_soon(IOLoop.current)\n            loop.call_soon(loop.stop)\n            loop.run_forever()\n            loop.close()\n        new_count = len(IOLoop._ioloop_for_asyncio) - orig_count\n        # Because the cleanup is run on new loop creation, we have one\n        # dangling entry in the map (but only one).\n        self.assertEqual(new_count, 1)\n\n\nclass AnyThreadEventLoopPolicyTest(unittest.TestCase):\n    def setUp(self):\n        self.orig_policy = asyncio.get_event_loop_policy()\n        self.executor = ThreadPoolExecutor(1)\n\n    def tearDown(self):\n        asyncio.set_event_loop_policy(self.orig_policy)\n        self.executor.shutdown()\n\n    def get_event_loop_on_thread(self):\n        def get_and_close_event_loop():\n            \"\"\"Get the event loop. Close it if one is returned.\n\n            Returns the (closed) event loop. This is a silly thing\n            to do and leaves the thread in a broken state, but it's\n            enough for this test. Closing the loop avoids resource\n            leak warnings.\n            \"\"\"\n            loop = asyncio.get_event_loop()\n            loop.close()\n            return loop\n\n        future = self.executor.submit(get_and_close_event_loop)\n        return future.result()\n\n    def run_policy_test(self, accessor, expected_type):\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            # With the default policy, non-main threads don't get an event\n            # loop.\n            self.assertRaises(\n                (RuntimeError, AssertionError), self.executor.submit(accessor).result\n            )\n            # Set the policy and we can get a loop.\n            asyncio.set_event_loop_policy(AnyThreadEventLoopPolicy())\n            self.assertIsInstance(\n                self.executor.submit(accessor).result(), expected_type\n            )\n            # Clean up to silence leak warnings. Always use asyncio since\n            # IOLoop doesn't (currently) close the underlying loop.\n            self.executor.submit(lambda: asyncio.get_event_loop().close()).result()  # type: ignore\n\n    def test_asyncio_accessor(self):\n        self.run_policy_test(asyncio.get_event_loop, asyncio.AbstractEventLoop)\n\n    def test_tornado_accessor(self):\n        self.run_policy_test(IOLoop.current, IOLoop)\n",
            "new_file_content": "# Licensed under the Apache License, Version 2.0 (the \"License\"); you may\n# not use this file except in compliance with the License. You may obtain\n# a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT\n# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n# License for the specific language governing permissions and limitations\n# under the License.\n\nimport asyncio\nimport unittest\nimport warnings\n\nfrom concurrent.futures import ThreadPoolExecutor\nfrom tornado import gen\nfrom tornado.ioloop import IOLoop\nfrom tornado.platform.asyncio import (\n    AsyncIOLoop,\n    to_asyncio_future,\n    AnyThreadEventLoopPolicy,\n)\nfrom tornado.testing import AsyncTestCase, gen_test\n\n\nclass AsyncIOLoopTest(AsyncTestCase):\n    def get_new_ioloop(self):\n        io_loop = AsyncIOLoop(make_current=False)\n        return io_loop\n\n    @property\n    def asyncio_loop(self):\n        return self.io_loop.asyncio_loop\n\n    def test_asyncio_callback(self):\n        # Basic test that the asyncio loop is set up correctly.\n        async def add_callback():\n            asyncio.get_event_loop().call_soon(self.stop)\n\n        self.asyncio_loop.run_until_complete(add_callback())\n        self.wait()\n\n    @gen_test\n    def test_asyncio_future(self):\n        # Test that we can yield an asyncio future from a tornado coroutine.\n        # Without 'yield from', we must wrap coroutines in ensure_future,\n        # which was introduced during Python 3.4, deprecating the prior \"async\".\n        if hasattr(asyncio, \"ensure_future\"):\n            ensure_future = asyncio.ensure_future\n        else:\n            # async is a reserved word in Python 3.7\n            ensure_future = getattr(asyncio, \"async\")\n\n        x = yield ensure_future(\n            asyncio.get_event_loop().run_in_executor(None, lambda: 42)\n        )\n        self.assertEqual(x, 42)\n\n    @gen_test\n    def test_asyncio_yield_from(self):\n        @gen.coroutine\n        def f():\n            event_loop = asyncio.get_event_loop()\n            x = yield from event_loop.run_in_executor(None, lambda: 42)\n            return x\n\n        result = yield f()\n        self.assertEqual(result, 42)\n\n    def test_asyncio_adapter(self):\n        # This test demonstrates that when using the asyncio coroutine\n        # runner (i.e. run_until_complete), the to_asyncio_future\n        # adapter is needed. No adapter is needed in the other direction,\n        # as demonstrated by other tests in the package.\n        @gen.coroutine\n        def tornado_coroutine():\n            yield gen.moment\n            raise gen.Return(42)\n\n        async def native_coroutine_without_adapter():\n            return await tornado_coroutine()\n\n        async def native_coroutine_with_adapter():\n            return await to_asyncio_future(tornado_coroutine())\n\n        # Use the adapter, but two degrees from the tornado coroutine.\n        async def native_coroutine_with_adapter2():\n            return await to_asyncio_future(native_coroutine_without_adapter())\n\n        # Tornado supports native coroutines both with and without adapters\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_without_adapter), 42)\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_with_adapter), 42)\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_with_adapter2), 42)\n\n        # Asyncio only supports coroutines that yield asyncio-compatible\n        # Futures (which our Future is since 5.0).\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_without_adapter()),\n            42,\n        )\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_with_adapter()),\n            42,\n        )\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_with_adapter2()),\n            42,\n        )\n        # I'm not entirely sure why this manual cleanup is necessary but without\n        # it we have at-a-distance failures in ioloop_test.TestIOLoopCurrent.\n        asyncio.set_event_loop(None)\n\n\nclass LeakTest(unittest.TestCase):\n    def setUp(self):\n        # Trigger a cleanup of the mapping so we start with a clean slate.\n        AsyncIOLoop(make_current=False).close()\n        # If we don't clean up after ourselves other tests may fail on\n        # py34.\n        self.orig_policy = asyncio.get_event_loop_policy()\n        asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())\n\n    def tearDown(self):\n        try:\n            loop = asyncio.get_event_loop_policy().get_event_loop()\n        except Exception:\n            # We may not have a current event loop at this point.\n            pass\n        else:\n            loop.close()\n        asyncio.set_event_loop_policy(self.orig_policy)\n\n    def test_ioloop_close_leak(self):\n        orig_count = len(IOLoop._ioloop_for_asyncio)\n        for i in range(10):\n            # Create and close an AsyncIOLoop using Tornado interfaces.\n            with warnings.catch_warnings():\n                warnings.simplefilter(\"ignore\", DeprecationWarning)\n                loop = AsyncIOLoop()\n                loop.close()\n        new_count = len(IOLoop._ioloop_for_asyncio) - orig_count\n        self.assertEqual(new_count, 0)\n\n    def test_asyncio_close_leak(self):\n        orig_count = len(IOLoop._ioloop_for_asyncio)\n        for i in range(10):\n            # Create and close an AsyncIOMainLoop using asyncio interfaces.\n            loop = asyncio.new_event_loop()\n            loop.call_soon(IOLoop.current)\n            loop.call_soon(loop.stop)\n            loop.run_forever()\n            loop.close()\n        new_count = len(IOLoop._ioloop_for_asyncio) - orig_count\n        # Because the cleanup is run on new loop creation, we have one\n        # dangling entry in the map (but only one).\n        self.assertEqual(new_count, 1)\n\n\nclass AnyThreadEventLoopPolicyTest(unittest.TestCase):\n    def setUp(self):\n        self.orig_policy = asyncio.get_event_loop_policy()\n        self.executor = ThreadPoolExecutor(1)\n\n    def tearDown(self):\n        asyncio.set_event_loop_policy(self.orig_policy)\n        self.executor.shutdown()\n\n    def get_event_loop_on_thread(self):\n        def get_and_close_event_loop():\n            \"\"\"Get the event loop. Close it if one is returned.\n\n            Returns the (closed) event loop. This is a silly thing\n            to do and leaves the thread in a broken state, but it's\n            enough for this test. Closing the loop avoids resource\n            leak warnings.\n            \"\"\"\n            loop = asyncio.get_event_loop()\n            loop.close()\n            return loop\n\n        future = self.executor.submit(get_and_close_event_loop)\n        return future.result()\n\n    def run_policy_test(self, accessor, expected_type):\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            # With the default policy, non-main threads don't get an event\n            # loop.\n            self.assertRaises(\n                (RuntimeError, AssertionError), self.executor.submit(accessor).result\n            )\n            # Set the policy and we can get a loop.\n            asyncio.set_event_loop_policy(AnyThreadEventLoopPolicy())\n            self.assertIsInstance(\n                self.executor.submit(accessor).result(), expected_type\n            )\n            # Clean up to silence leak warnings. Always use asyncio since\n            # IOLoop doesn't (currently) close the underlying loop.\n            self.executor.submit(lambda: asyncio.get_event_loop().close()).result()  # type: ignore\n\n    def test_asyncio_accessor(self):\n        self.run_policy_test(asyncio.get_event_loop, asyncio.AbstractEventLoop)\n\n    def test_tornado_accessor(self):\n        self.run_policy_test(IOLoop.current, IOLoop)\n",
            "header": {
                "file": {
                    "path": "tornado/test/asyncio_test.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "c33d7256",
                "new_commit_hash": "6ad1dcfb",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/tornado/test/asyncio_test.py"
            },
            "plus_file": {
                "path": "b/tornado/test/asyncio_test.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 30,
                            "length": 6
                        },
                        "new_range": {
                            "start": 30,
                            "length": 10
                        },
                        "section": "class AsyncIOLoopTest(AsyncTestCase):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        io_loop = AsyncIOLoop(make_current=False)",
                                "type": "context"
                            },
                            {
                                "content": "        return io_loop",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    @property",
                                "type": "added"
                            },
                            {
                                "content": "    def asyncio_loop(self):",
                                "type": "added"
                            },
                            {
                                "content": "        return self.io_loop.asyncio_loop",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "    def test_asyncio_callback(self):",
                                "type": "context"
                            },
                            {
                                "content": "        # Basic test that the asyncio loop is set up correctly.",
                                "type": "context"
                            },
                            {
                                "content": "        async def add_callback():",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/test/asyncio_test.py",
                            "type": "class",
                            "name": "AsyncIOLoopTest",
                            "content": "class AsyncIOLoopTest(AsyncTestCase):\n    def get_new_ioloop(self):\n        io_loop = AsyncIOLoop(make_current=False)\n        return io_loop\n\n    @property\n    def asyncio_loop(self):\n        return self.io_loop.asyncio_loop\n\n    def test_asyncio_callback(self):\n        # Basic test that the asyncio loop is set up correctly.\n        async def add_callback():\n            asyncio.get_event_loop().call_soon(self.stop)\n\n        self.asyncio_loop.run_until_complete(add_callback())\n        self.wait()\n\n    @gen_test\n    def test_asyncio_future(self):\n        # Test that we can yield an asyncio future from a tornado coroutine.\n        # Without 'yield from', we must wrap coroutines in ensure_future,\n        # which was introduced during Python 3.4, deprecating the prior \"async\".\n        if hasattr(asyncio, \"ensure_future\"):\n            ensure_future = asyncio.ensure_future\n        else:\n            # async is a reserved word in Python 3.7\n            ensure_future = getattr(asyncio, \"async\")\n\n        x = yield ensure_future(\n            asyncio.get_event_loop().run_in_executor(None, lambda: 42)\n        )\n        self.assertEqual(x, 42)\n\n    @gen_test\n    def test_asyncio_yield_from(self):\n        @gen.coroutine\n        def f():\n            event_loop = asyncio.get_event_loop()\n            x = yield from event_loop.run_in_executor(None, lambda: 42)\n            return x\n\n        result = yield f()\n        self.assertEqual(result, 42)\n\n    def test_asyncio_adapter(self):\n        # This test demonstrates that when using the asyncio coroutine\n        # runner (i.e. run_until_complete), the to_asyncio_future\n        # adapter is needed. No adapter is needed in the other direction,\n        # as demonstrated by other tests in the package.\n        @gen.coroutine\n        def tornado_coroutine():\n            yield gen.moment\n            raise gen.Return(42)\n\n        async def native_coroutine_without_adapter():\n            return await tornado_coroutine()\n\n        async def native_coroutine_with_adapter():\n            return await to_asyncio_future(tornado_coroutine())\n\n        # Use the adapter, but two degrees from the tornado coroutine.\n        async def native_coroutine_with_adapter2():\n            return await to_asyncio_future(native_coroutine_without_adapter())\n\n        # Tornado supports native coroutines both with and without adapters\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_without_adapter), 42)\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_with_adapter), 42)\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_with_adapter2), 42)\n\n        # Asyncio only supports coroutines that yield asyncio-compatible\n        # Futures (which our Future is since 5.0).\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_without_adapter()),\n            42,\n        )\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_with_adapter()),\n            42,\n        )\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_with_adapter2()),\n            42,\n        )\n        # I'm not entirely sure why this manual cleanup is necessary but without\n        # it we have at-a-distance failures in ioloop_test.TestIOLoopCurrent.\n        asyncio.set_event_loop(None)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 28,
                            "end_lineno": 113,
                            "parent": null
                        }
                    ],
                    "added_entities": [
                        {
                            "file_name": "tornado/test/asyncio_test.py",
                            "type": "method",
                            "name": "AsyncIOLoopTest.asyncio_loop",
                            "content": "class AsyncIOLoopTest(AsyncTestCase):\n    def get_new_ioloop(self):\n        io_loop = AsyncIOLoop(make_current=False)\n        return io_loop\n\n    @property\n    def asyncio_loop(self):\n        return self.io_loop.asyncio_loop\n\n    def test_asyncio_callback(self):\n        # Basic test that the asyncio loop is set up correctly.\n        async def add_callback():\n            asyncio.get_event_loop().call_soon(self.stop)\n\n        self.asyncio_loop.run_until_complete(add_callback())\n        self.wait()\n\n    @gen_test\n    def test_asyncio_future(self):\n        # Test that we can yield an asyncio future from a tornado coroutine.\n        # Without 'yield from', we must wrap coroutines in ensure_future,\n        # which was introduced during Python 3.4, deprecating the prior \"async\".\n        if hasattr(asyncio, \"ensure_future\"):\n            ensure_future = asyncio.ensure_future\n        else:\n            # async is a reserved word in Python 3.7\n            ensure_future = getattr(asyncio, \"async\")\n\n        x = yield ensure_future(\n            asyncio.get_event_loop().run_in_executor(None, lambda: 42)\n        )\n        self.assertEqual(x, 42)\n\n    @gen_test\n    def test_asyncio_yield_from(self):\n        @gen.coroutine\n        def f():\n            event_loop = asyncio.get_event_loop()\n            x = yield from event_loop.run_in_executor(None, lambda: 42)\n            return x\n\n        result = yield f()\n        self.assertEqual(result, 42)\n\n    def test_asyncio_adapter(self):\n        # This test demonstrates that when using the asyncio coroutine\n        # runner (i.e. run_until_complete), the to_asyncio_future\n        # adapter is needed. No adapter is needed in the other direction,\n        # as demonstrated by other tests in the package.\n        @gen.coroutine\n        def tornado_coroutine():\n            yield gen.moment\n            raise gen.Return(42)\n\n        async def native_coroutine_without_adapter():\n            return await tornado_coroutine()\n\n        async def native_coroutine_with_adapter():\n            return await to_asyncio_future(tornado_coroutine())\n\n        # Use the adapter, but two degrees from the tornado coroutine.\n        async def native_coroutine_with_adapter2():\n            return await to_asyncio_future(native_coroutine_without_adapter())\n\n        # Tornado supports native coroutines both with and without adapters\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_without_adapter), 42)\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_with_adapter), 42)\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_with_adapter2), 42)\n\n        # Asyncio only supports coroutines that yield asyncio-compatible\n        # Futures (which our Future is since 5.0).\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_without_adapter()),\n            42,\n        )\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_with_adapter()),\n            42,\n        )\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_with_adapter2()),\n            42,\n        )\n        # I'm not entirely sure why this manual cleanup is necessary but without\n        # it we have at-a-distance failures in ioloop_test.TestIOLoopCurrent.\n        asyncio.set_event_loop(None)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 34,
                            "end_lineno": 35,
                            "parent": {
                                "file_name": "tornado/test/asyncio_test.py",
                                "type": "class",
                                "name": "AsyncIOLoopTest",
                                "content": "class AsyncIOLoopTest(AsyncTestCase):\n    def get_new_ioloop(self):\n        io_loop = AsyncIOLoop(make_current=False)\n        return io_loop\n\n    @property\n    def asyncio_loop(self):\n        return self.io_loop.asyncio_loop\n\n    def test_asyncio_callback(self):\n        # Basic test that the asyncio loop is set up correctly.\n        async def add_callback():\n            asyncio.get_event_loop().call_soon(self.stop)\n\n        self.asyncio_loop.run_until_complete(add_callback())\n        self.wait()\n\n    @gen_test\n    def test_asyncio_future(self):\n        # Test that we can yield an asyncio future from a tornado coroutine.\n        # Without 'yield from', we must wrap coroutines in ensure_future,\n        # which was introduced during Python 3.4, deprecating the prior \"async\".\n        if hasattr(asyncio, \"ensure_future\"):\n            ensure_future = asyncio.ensure_future\n        else:\n            # async is a reserved word in Python 3.7\n            ensure_future = getattr(asyncio, \"async\")\n\n        x = yield ensure_future(\n            asyncio.get_event_loop().run_in_executor(None, lambda: 42)\n        )\n        self.assertEqual(x, 42)\n\n    @gen_test\n    def test_asyncio_yield_from(self):\n        @gen.coroutine\n        def f():\n            event_loop = asyncio.get_event_loop()\n            x = yield from event_loop.run_in_executor(None, lambda: 42)\n            return x\n\n        result = yield f()\n        self.assertEqual(result, 42)\n\n    def test_asyncio_adapter(self):\n        # This test demonstrates that when using the asyncio coroutine\n        # runner (i.e. run_until_complete), the to_asyncio_future\n        # adapter is needed. No adapter is needed in the other direction,\n        # as demonstrated by other tests in the package.\n        @gen.coroutine\n        def tornado_coroutine():\n            yield gen.moment\n            raise gen.Return(42)\n\n        async def native_coroutine_without_adapter():\n            return await tornado_coroutine()\n\n        async def native_coroutine_with_adapter():\n            return await to_asyncio_future(tornado_coroutine())\n\n        # Use the adapter, but two degrees from the tornado coroutine.\n        async def native_coroutine_with_adapter2():\n            return await to_asyncio_future(native_coroutine_without_adapter())\n\n        # Tornado supports native coroutines both with and without adapters\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_without_adapter), 42)\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_with_adapter), 42)\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_with_adapter2), 42)\n\n        # Asyncio only supports coroutines that yield asyncio-compatible\n        # Futures (which our Future is since 5.0).\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_without_adapter()),\n            42,\n        )\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_with_adapter()),\n            42,\n        )\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_with_adapter2()),\n            42,\n        )\n        # I'm not entirely sure why this manual cleanup is necessary but without\n        # it we have at-a-distance failures in ioloop_test.TestIOLoopCurrent.\n        asyncio.set_event_loop(None)",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 28,
                                "end_lineno": 113,
                                "parent": null
                            }
                        }
                    ],
                    "deleted_entities": []
                }
            ]
        },
        {
            "old_file_content": "\"\"\"Support classes for automated testing.\n\n* `AsyncTestCase` and `AsyncHTTPTestCase`:  Subclasses of unittest.TestCase\n  with additional support for testing asynchronous (`.IOLoop`-based) code.\n\n* `ExpectLog`: Make test logs less spammy.\n\n* `main()`: A simple test runner (wrapper around unittest.main()) with support\n  for the tornado.autoreload module to rerun the tests when code changes.\n\"\"\"\n\nimport asyncio\nfrom collections.abc import Generator\nimport functools\nimport inspect\nimport logging\nimport os\nimport re\nimport signal\nimport socket\nimport sys\nimport unittest\nimport warnings\n\nfrom tornado import gen\nfrom tornado.httpclient import AsyncHTTPClient, HTTPResponse\nfrom tornado.httpserver import HTTPServer\nfrom tornado.ioloop import IOLoop, TimeoutError\nfrom tornado import netutil\nfrom tornado.platform.asyncio import AsyncIOMainLoop\nfrom tornado.process import Subprocess\nfrom tornado.log import app_log\nfrom tornado.util import raise_exc_info, basestring_type\nfrom tornado.web import Application\n\nimport typing\nfrom typing import Tuple, Any, Callable, Type, Dict, Union, Optional, Coroutine\nfrom types import TracebackType\n\nif typing.TYPE_CHECKING:\n    _ExcInfoTuple = Tuple[\n        Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]\n    ]\n\n\n_NON_OWNED_IOLOOPS = AsyncIOMainLoop\n\n\ndef bind_unused_port(\n    reuse_port: bool = False, address: str = \"127.0.0.1\"\n) -> Tuple[socket.socket, int]:\n    \"\"\"Binds a server socket to an available port on localhost.\n\n    Returns a tuple (socket, port).\n\n    .. versionchanged:: 4.4\n       Always binds to ``127.0.0.1`` without resolving the name\n       ``localhost``.\n\n    .. versionchanged:: 6.2\n       Added optional ``address`` argument to\n       override the default \"127.0.0.1\".\n    \"\"\"\n    sock = netutil.bind_sockets(\n        0, address, family=socket.AF_INET, reuse_port=reuse_port\n    )[0]\n    port = sock.getsockname()[1]\n    return sock, port\n\n\ndef get_async_test_timeout() -> float:\n    \"\"\"Get the global timeout setting for async tests.\n\n    Returns a float, the timeout in seconds.\n\n    .. versionadded:: 3.1\n    \"\"\"\n    env = os.environ.get(\"ASYNC_TEST_TIMEOUT\")\n    if env is not None:\n        try:\n            return float(env)\n        except ValueError:\n            pass\n    return 5\n\n\nclass _TestMethodWrapper(object):\n    \"\"\"Wraps a test method to raise an error if it returns a value.\n\n    This is mainly used to detect undecorated generators (if a test\n    method yields it must use a decorator to consume the generator),\n    but will also detect other kinds of return values (these are not\n    necessarily errors, but we alert anyway since there is no good\n    reason to return a value from a test).\n    \"\"\"\n\n    def __init__(self, orig_method: Callable) -> None:\n        self.orig_method = orig_method\n        self.__wrapped__ = orig_method\n\n    def __call__(self, *args: Any, **kwargs: Any) -> None:\n        result = self.orig_method(*args, **kwargs)\n        if isinstance(result, Generator) or inspect.iscoroutine(result):\n            raise TypeError(\n                \"Generator and coroutine test methods should be\"\n                \" decorated with tornado.testing.gen_test\"\n            )\n        elif result is not None:\n            raise ValueError(\"Return value from test method ignored: %r\" % result)\n\n    def __getattr__(self, name: str) -> Any:\n        \"\"\"Proxy all unknown attributes to the original method.\n\n        This is important for some of the decorators in the `unittest`\n        module, such as `unittest.skipIf`.\n        \"\"\"\n        return getattr(self.orig_method, name)\n\n\nclass AsyncTestCase(unittest.TestCase):\n    \"\"\"`~unittest.TestCase` subclass for testing `.IOLoop`-based\n    asynchronous code.\n\n    The unittest framework is synchronous, so the test must be\n    complete by the time the test method returns. This means that\n    asynchronous code cannot be used in quite the same way as usual\n    and must be adapted to fit. To write your tests with coroutines,\n    decorate your test methods with `tornado.testing.gen_test` instead\n    of `tornado.gen.coroutine`.\n\n    This class also provides the (deprecated) `stop()` and `wait()`\n    methods for a more manual style of testing. The test method itself\n    must call ``self.wait()``, and asynchronous callbacks should call\n    ``self.stop()`` to signal completion.\n\n    By default, a new `.IOLoop` is constructed for each test and is available\n    as ``self.io_loop``.  If the code being tested requires a\n    global `.IOLoop`, subclasses should override `get_new_ioloop` to return it.\n\n    The `.IOLoop`'s ``start`` and ``stop`` methods should not be\n    called directly.  Instead, use `self.stop <stop>` and `self.wait\n    <wait>`.  Arguments passed to ``self.stop`` are returned from\n    ``self.wait``.  It is possible to have multiple ``wait``/``stop``\n    cycles in the same test.\n\n    Example::\n\n        # This test uses coroutine style.\n        class MyTestCase(AsyncTestCase):\n            @tornado.testing.gen_test\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                response = yield client.fetch(\"http://www.tornadoweb.org\")\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n\n        # This test uses argument passing between self.stop and self.wait.\n        class MyTestCase2(AsyncTestCase):\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                client.fetch(\"http://www.tornadoweb.org/\", self.stop)\n                response = self.wait()\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n    \"\"\"\n\n    def __init__(self, methodName: str = \"runTest\") -> None:\n        super().__init__(methodName)\n        self.__stopped = False\n        self.__running = False\n        self.__failure = None  # type: Optional[_ExcInfoTuple]\n        self.__stop_args = None  # type: Any\n        self.__timeout = None  # type: Optional[object]\n\n        # It's easy to forget the @gen_test decorator, but if you do\n        # the test will silently be ignored because nothing will consume\n        # the generator.  Replace the test method with a wrapper that will\n        # make sure it's not an undecorated generator.\n        setattr(self, methodName, _TestMethodWrapper(getattr(self, methodName)))\n\n        # Not used in this class itself, but used by @gen_test\n        self._test_generator = None  # type: Optional[Union[Generator, Coroutine]]\n\n    def setUp(self) -> None:\n        setup_with_context_manager(self, warnings.catch_warnings())\n        warnings.filterwarnings(\n            \"ignore\",\n            message=\"There is no current event loop\",\n            category=DeprecationWarning,\n            module=r\"tornado\\..*\",\n        )\n        super().setUp()\n        # NOTE: this code attempts to navigate deprecation warnings introduced\n        # in Python 3.10. The idea of an implicit current event loop is\n        # deprecated in that version, with the intention that tests like this\n        # explicitly create a new event loop and run on it. However, other\n        # packages such as pytest-asyncio (as of version 0.16.0) still rely on\n        # the implicit current event loop and we want to be compatible with them\n        # (even when run on 3.10, but not, of course, on the future version of\n        # python that removes the get/set_event_loop methods completely).\n        #\n        # Deprecation warnings were introduced inconsistently:\n        # asyncio.get_event_loop warns, but\n        # asyncio.get_event_loop_policy().get_event_loop does not. Similarly,\n        # none of the set_event_loop methods warn, although comments on\n        # https://bugs.python.org/issue39529 indicate that they are also\n        # intended for future removal.\n        #\n        # Therefore, we first attempt to access the event loop with the\n        # (non-warning) policy method, and if it fails, fall back to creating a\n        # new event loop. We do not have effective test coverage of the\n        # new event loop case; this will have to be watched when/if\n        # get_event_loop is actually removed.\n        self.should_close_asyncio_loop = False\n        try:\n            self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()\n        except Exception:\n            self.asyncio_loop = asyncio.new_event_loop()\n            self.should_close_asyncio_loop = True\n\n        async def get_loop() -> IOLoop:\n            return self.get_new_ioloop()\n\n        self.io_loop = self.asyncio_loop.run_until_complete(get_loop())\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.make_current()\n\n    def tearDown(self) -> None:\n        # Native coroutines tend to produce warnings if they're not\n        # allowed to run to completion. It's difficult to ensure that\n        # this always happens in tests, so cancel any tasks that are\n        # still pending by the time we get here.\n        asyncio_loop = self.io_loop.asyncio_loop  # type: ignore\n        if hasattr(asyncio, \"all_tasks\"):  # py37\n            tasks = asyncio.all_tasks(asyncio_loop)  # type: ignore\n        else:\n            tasks = asyncio.Task.all_tasks(asyncio_loop)\n        # Tasks that are done may still appear here and may contain\n        # non-cancellation exceptions, so filter them out.\n        tasks = [t for t in tasks if not t.done()]  # type: ignore\n        for t in tasks:\n            t.cancel()\n        # Allow the tasks to run and finalize themselves (which means\n        # raising a CancelledError inside the coroutine). This may\n        # just transform the \"task was destroyed but it is pending\"\n        # warning into a \"uncaught CancelledError\" warning, but\n        # catching CancelledErrors in coroutines that may leak is\n        # simpler than ensuring that no coroutines leak.\n        if tasks:\n            done, pending = self.io_loop.run_sync(lambda: asyncio.wait(tasks))\n            assert not pending\n            # If any task failed with anything but a CancelledError, raise it.\n            for f in done:\n                try:\n                    f.result()\n                except asyncio.CancelledError:\n                    pass\n\n        # Clean up Subprocess, so it can be used again with a new ioloop.\n        Subprocess.uninitialize()\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.clear_current()\n        if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):\n            # Try to clean up any file descriptors left open in the ioloop.\n            # This avoids leaks, especially when tests are run repeatedly\n            # in the same process with autoreload (because curl does not\n            # set FD_CLOEXEC on its file descriptors)\n            self.io_loop.close(all_fds=True)\n        if self.should_close_asyncio_loop:\n            self.asyncio_loop.close()\n        super().tearDown()\n        # In case an exception escaped or the StackContext caught an exception\n        # when there wasn't a wait() to re-raise it, do so here.\n        # This is our last chance to raise an exception in a way that the\n        # unittest machinery understands.\n        self.__rethrow()\n\n    def get_new_ioloop(self) -> IOLoop:\n        \"\"\"Returns the `.IOLoop` to use for this test.\n\n        By default, a new `.IOLoop` is created for each test.\n        Subclasses may override this method to return\n        `.IOLoop.current()` if it is not appropriate to use a new\n        `.IOLoop` in each tests (for example, if there are global\n        singletons using the default `.IOLoop`) or if a per-test event\n        loop is being provided by another system (such as\n        ``pytest-asyncio``).\n        \"\"\"\n        return IOLoop(make_current=False)\n\n    def _handle_exception(\n        self, typ: Type[Exception], value: Exception, tb: TracebackType\n    ) -> bool:\n        if self.__failure is None:\n            self.__failure = (typ, value, tb)\n        else:\n            app_log.error(\n                \"multiple unhandled exceptions in test\", exc_info=(typ, value, tb)\n            )\n        self.stop()\n        return True\n\n    def __rethrow(self) -> None:\n        if self.__failure is not None:\n            failure = self.__failure\n            self.__failure = None\n            raise_exc_info(failure)\n\n    def run(\n        self, result: Optional[unittest.TestResult] = None\n    ) -> Optional[unittest.TestResult]:\n        ret = super().run(result)\n        # As a last resort, if an exception escaped super.run() and wasn't\n        # re-raised in tearDown, raise it here.  This will cause the\n        # unittest run to fail messily, but that's better than silently\n        # ignoring an error.\n        self.__rethrow()\n        return ret\n\n    def stop(self, _arg: Any = None, **kwargs: Any) -> None:\n        \"\"\"Stops the `.IOLoop`, causing one pending (or future) call to `wait()`\n        to return.\n\n        Keyword arguments or a single positional argument passed to `stop()` are\n        saved and will be returned by `wait()`.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        assert _arg is None or not kwargs\n        self.__stop_args = kwargs or _arg\n        if self.__running:\n            self.io_loop.stop()\n            self.__running = False\n        self.__stopped = True\n\n    def wait(\n        self,\n        condition: Optional[Callable[..., bool]] = None,\n        timeout: Optional[float] = None,\n    ) -> Any:\n        \"\"\"Runs the `.IOLoop` until stop is called or timeout has passed.\n\n        In the event of a timeout, an exception will be thrown. The\n        default timeout is 5 seconds; it may be overridden with a\n        ``timeout`` keyword argument or globally with the\n        ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        If ``condition`` is not ``None``, the `.IOLoop` will be restarted\n        after `stop()` until ``condition()`` returns ``True``.\n\n        .. versionchanged:: 3.1\n           Added the ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        if timeout is None:\n            timeout = get_async_test_timeout()\n\n        if not self.__stopped:\n            if timeout:\n\n                def timeout_func() -> None:\n                    try:\n                        raise self.failureException(\n                            \"Async operation timed out after %s seconds\" % timeout\n                        )\n                    except Exception:\n                        self.__failure = sys.exc_info()\n                    self.stop()\n\n                self.__timeout = self.io_loop.add_timeout(\n                    self.io_loop.time() + timeout, timeout_func\n                )\n            while True:\n                self.__running = True\n                self.io_loop.start()\n                if self.__failure is not None or condition is None or condition():\n                    break\n            if self.__timeout is not None:\n                self.io_loop.remove_timeout(self.__timeout)\n                self.__timeout = None\n        assert self.__stopped\n        self.__stopped = False\n        self.__rethrow()\n        result = self.__stop_args\n        self.__stop_args = None\n        return result\n\n\nclass AsyncHTTPTestCase(AsyncTestCase):\n    \"\"\"A test case that starts up an HTTP server.\n\n    Subclasses must override `get_app()`, which returns the\n    `tornado.web.Application` (or other `.HTTPServer` callback) to be tested.\n    Tests will typically use the provided ``self.http_client`` to fetch\n    URLs from this server.\n\n    Example, assuming the \"Hello, world\" example from the user guide is in\n    ``hello.py``::\n\n        import hello\n\n        class TestHelloApp(AsyncHTTPTestCase):\n            def get_app(self):\n                return hello.make_app()\n\n            def test_homepage(self):\n                response = self.fetch('/')\n                self.assertEqual(response.code, 200)\n                self.assertEqual(response.body, 'Hello, world')\n\n    That call to ``self.fetch()`` is equivalent to ::\n\n        self.http_client.fetch(self.get_url('/'), self.stop)\n        response = self.wait()\n\n    which illustrates how AsyncTestCase can turn an asynchronous operation,\n    like ``http_client.fetch()``, into a synchronous operation. If you need\n    to do other asynchronous operations in tests, you'll probably need to use\n    ``stop()`` and ``wait()`` yourself.\n    \"\"\"\n\n    def setUp(self) -> None:\n        super().setUp()\n        sock, port = bind_unused_port()\n        self.__port = port\n\n        self.http_client = self.get_http_client()\n        self._app = self.get_app()\n        self.http_server = self.get_http_server()\n        self.http_server.add_sockets([sock])\n\n    def get_http_client(self) -> AsyncHTTPClient:\n        return AsyncHTTPClient()\n\n    def get_http_server(self) -> HTTPServer:\n        return HTTPServer(self._app, **self.get_httpserver_options())\n\n    def get_app(self) -> Application:\n        \"\"\"Should be overridden by subclasses to return a\n        `tornado.web.Application` or other `.HTTPServer` callback.\n        \"\"\"\n        raise NotImplementedError()\n\n    def fetch(\n        self, path: str, raise_error: bool = False, **kwargs: Any\n    ) -> HTTPResponse:\n        \"\"\"Convenience method to synchronously fetch a URL.\n\n        The given path will be appended to the local server's host and\n        port.  Any additional keyword arguments will be passed directly to\n        `.AsyncHTTPClient.fetch` (and so could be used to pass\n        ``method=\"POST\"``, ``body=\"...\"``, etc).\n\n        If the path begins with http:// or https://, it will be treated as a\n        full URL and will be fetched as-is.\n\n        If ``raise_error`` is ``True``, a `tornado.httpclient.HTTPError` will\n        be raised if the response code is not 200. This is the same behavior\n        as the ``raise_error`` argument to `.AsyncHTTPClient.fetch`, but\n        the default is ``False`` here (it's ``True`` in `.AsyncHTTPClient`)\n        because tests often need to deal with non-200 response codes.\n\n        .. versionchanged:: 5.0\n           Added support for absolute URLs.\n\n        .. versionchanged:: 5.1\n\n           Added the ``raise_error`` argument.\n\n        .. deprecated:: 5.1\n\n           This method currently turns any exception into an\n           `.HTTPResponse` with status code 599. In Tornado 6.0,\n           errors other than `tornado.httpclient.HTTPError` will be\n           passed through, and ``raise_error=False`` will only\n           suppress errors that would be raised due to non-200\n           response codes.\n\n        \"\"\"\n        if path.lower().startswith((\"http://\", \"https://\")):\n            url = path\n        else:\n            url = self.get_url(path)\n        return self.io_loop.run_sync(\n            lambda: self.http_client.fetch(url, raise_error=raise_error, **kwargs),\n            timeout=get_async_test_timeout(),\n        )\n\n    def get_httpserver_options(self) -> Dict[str, Any]:\n        \"\"\"May be overridden by subclasses to return additional\n        keyword arguments for the server.\n        \"\"\"\n        return {}\n\n    def get_http_port(self) -> int:\n        \"\"\"Returns the port used by the server.\n\n        A new port is chosen for each test.\n        \"\"\"\n        return self.__port\n\n    def get_protocol(self) -> str:\n        return \"http\"\n\n    def get_url(self, path: str) -> str:\n        \"\"\"Returns an absolute url for the given path on the test server.\"\"\"\n        return \"%s://127.0.0.1:%s%s\" % (self.get_protocol(), self.get_http_port(), path)\n\n    def tearDown(self) -> None:\n        self.http_server.stop()\n        self.io_loop.run_sync(\n            self.http_server.close_all_connections, timeout=get_async_test_timeout()\n        )\n        self.http_client.close()\n        del self.http_server\n        del self._app\n        super().tearDown()\n\n\nclass AsyncHTTPSTestCase(AsyncHTTPTestCase):\n    \"\"\"A test case that starts an HTTPS server.\n\n    Interface is generally the same as `AsyncHTTPTestCase`.\n    \"\"\"\n\n    def get_http_client(self) -> AsyncHTTPClient:\n        return AsyncHTTPClient(force_instance=True, defaults=dict(validate_cert=False))\n\n    def get_httpserver_options(self) -> Dict[str, Any]:\n        return dict(ssl_options=self.get_ssl_options())\n\n    def get_ssl_options(self) -> Dict[str, Any]:\n        \"\"\"May be overridden by subclasses to select SSL options.\n\n        By default includes a self-signed testing certificate.\n        \"\"\"\n        return AsyncHTTPSTestCase.default_ssl_options()\n\n    @staticmethod\n    def default_ssl_options() -> Dict[str, Any]:\n        # Testing keys were generated with:\n        # openssl req -new -keyout tornado/test/test.key \\\n        #                     -out tornado/test/test.crt -nodes -days 3650 -x509\n        module_dir = os.path.dirname(__file__)\n        return dict(\n            certfile=os.path.join(module_dir, \"test\", \"test.crt\"),\n            keyfile=os.path.join(module_dir, \"test\", \"test.key\"),\n        )\n\n    def get_protocol(self) -> str:\n        return \"https\"\n\n\n@typing.overload\ndef gen_test(\n    *, timeout: Optional[float] = None\n) -> Callable[[Callable[..., Union[Generator, \"Coroutine\"]]], Callable[..., None]]:\n    pass\n\n\n@typing.overload  # noqa: F811\ndef gen_test(func: Callable[..., Union[Generator, \"Coroutine\"]]) -> Callable[..., None]:\n    pass\n\n\ndef gen_test(  # noqa: F811\n    func: Optional[Callable[..., Union[Generator, \"Coroutine\"]]] = None,\n    timeout: Optional[float] = None,\n) -> Union[\n    Callable[..., None],\n    Callable[[Callable[..., Union[Generator, \"Coroutine\"]]], Callable[..., None]],\n]:\n    \"\"\"Testing equivalent of ``@gen.coroutine``, to be applied to test methods.\n\n    ``@gen.coroutine`` cannot be used on tests because the `.IOLoop` is not\n    already running.  ``@gen_test`` should be applied to test methods\n    on subclasses of `AsyncTestCase`.\n\n    Example::\n\n        class MyTest(AsyncHTTPTestCase):\n            @gen_test\n            def test_something(self):\n                response = yield self.http_client.fetch(self.get_url('/'))\n\n    By default, ``@gen_test`` times out after 5 seconds. The timeout may be\n    overridden globally with the ``ASYNC_TEST_TIMEOUT`` environment variable,\n    or for each test with the ``timeout`` keyword argument::\n\n        class MyTest(AsyncHTTPTestCase):\n            @gen_test(timeout=10)\n            def test_something_slow(self):\n                response = yield self.http_client.fetch(self.get_url('/'))\n\n    Note that ``@gen_test`` is incompatible with `AsyncTestCase.stop`,\n    `AsyncTestCase.wait`, and `AsyncHTTPTestCase.fetch`. Use ``yield\n    self.http_client.fetch(self.get_url())`` as shown above instead.\n\n    .. versionadded:: 3.1\n       The ``timeout`` argument and ``ASYNC_TEST_TIMEOUT`` environment\n       variable.\n\n    .. versionchanged:: 4.0\n       The wrapper now passes along ``*args, **kwargs`` so it can be used\n       on functions with arguments.\n\n    \"\"\"\n    if timeout is None:\n        timeout = get_async_test_timeout()\n\n    def wrap(f: Callable[..., Union[Generator, \"Coroutine\"]]) -> Callable[..., None]:\n        # Stack up several decorators to allow us to access the generator\n        # object itself.  In the innermost wrapper, we capture the generator\n        # and save it in an attribute of self.  Next, we run the wrapped\n        # function through @gen.coroutine.  Finally, the coroutine is\n        # wrapped again to make it synchronous with run_sync.\n        #\n        # This is a good case study arguing for either some sort of\n        # extensibility in the gen decorators or cancellation support.\n        @functools.wraps(f)\n        def pre_coroutine(self, *args, **kwargs):\n            # type: (AsyncTestCase, *Any, **Any) -> Union[Generator, Coroutine]\n            # Type comments used to avoid pypy3 bug.\n            result = f(self, *args, **kwargs)\n            if isinstance(result, Generator) or inspect.iscoroutine(result):\n                self._test_generator = result\n            else:\n                self._test_generator = None\n            return result\n\n        if inspect.iscoroutinefunction(f):\n            coro = pre_coroutine\n        else:\n            coro = gen.coroutine(pre_coroutine)  # type: ignore[assignment]\n\n        @functools.wraps(coro)\n        def post_coroutine(self, *args, **kwargs):\n            # type: (AsyncTestCase, *Any, **Any) -> None\n            try:\n                return self.io_loop.run_sync(\n                    functools.partial(coro, self, *args, **kwargs), timeout=timeout\n                )\n            except TimeoutError as e:\n                # run_sync raises an error with an unhelpful traceback.\n                # If the underlying generator is still running, we can throw the\n                # exception back into it so the stack trace is replaced by the\n                # point where the test is stopped. The only reason the generator\n                # would not be running would be if it were cancelled, which means\n                # a native coroutine, so we can rely on the cr_running attribute.\n                if self._test_generator is not None and getattr(\n                    self._test_generator, \"cr_running\", True\n                ):\n                    self._test_generator.throw(e)\n                    # In case the test contains an overly broad except\n                    # clause, we may get back here.\n                # Coroutine was stopped or didn't raise a useful stack trace,\n                # so re-raise the original exception which is better than nothing.\n                raise\n\n        return post_coroutine\n\n    if func is not None:\n        # Used like:\n        #     @gen_test\n        #     def f(self):\n        #         pass\n        return wrap(func)\n    else:\n        # Used like @gen_test(timeout=10)\n        return wrap\n\n\n# Without this attribute, nosetests will try to run gen_test as a test\n# anywhere it is imported.\ngen_test.__test__ = False  # type: ignore\n\n\nclass ExpectLog(logging.Filter):\n    \"\"\"Context manager to capture and suppress expected log output.\n\n    Useful to make tests of error conditions less noisy, while still\n    leaving unexpected log entries visible.  *Not thread safe.*\n\n    The attribute ``logged_stack`` is set to ``True`` if any exception\n    stack trace was logged.\n\n    Usage::\n\n        with ExpectLog('tornado.application', \"Uncaught exception\"):\n            error_response = self.fetch(\"/some_page\")\n\n    .. versionchanged:: 4.3\n       Added the ``logged_stack`` attribute.\n    \"\"\"\n\n    def __init__(\n        self,\n        logger: Union[logging.Logger, basestring_type],\n        regex: str,\n        required: bool = True,\n        level: Optional[int] = None,\n    ) -> None:\n        \"\"\"Constructs an ExpectLog context manager.\n\n        :param logger: Logger object (or name of logger) to watch.  Pass an\n            empty string to watch the root logger.\n        :param regex: Regular expression to match.  Any log entries on the\n            specified logger that match this regex will be suppressed.\n        :param required: If true, an exception will be raised if the end of the\n            ``with`` statement is reached without matching any log entries.\n        :param level: A constant from the ``logging`` module indicating the\n            expected log level. If this parameter is provided, only log messages\n            at this level will be considered to match. Additionally, the\n            supplied ``logger`` will have its level adjusted if necessary (for\n            the duration of the ``ExpectLog`` to enable the expected message.\n\n        .. versionchanged:: 6.1\n           Added the ``level`` parameter.\n\n        .. deprecated:: 6.3\n           In Tornado 7.0, only ``WARNING`` and higher logging levels will be\n           matched by default. To match ``INFO`` and lower levels, the ``level``\n           argument must be used. This is changing to minimize differences\n           between ``tornado.testing.main`` (which enables ``INFO`` logs by\n           default) and most other test runners (including those in IDEs)\n           which have ``INFO`` logs disabled by default.\n        \"\"\"\n        if isinstance(logger, basestring_type):\n            logger = logging.getLogger(logger)\n        self.logger = logger\n        self.regex = re.compile(regex)\n        self.required = required\n        # matched and deprecated_level_matched are a counter for the respective event.\n        self.matched = 0\n        self.deprecated_level_matched = 0\n        self.logged_stack = False\n        self.level = level\n        self.orig_level = None  # type: Optional[int]\n\n    def filter(self, record: logging.LogRecord) -> bool:\n        if record.exc_info:\n            self.logged_stack = True\n        message = record.getMessage()\n        if self.regex.match(message):\n            if self.level is None and record.levelno < logging.WARNING:\n                # We're inside the logging machinery here so generating a DeprecationWarning\n                # here won't be reported cleanly (if warnings-as-errors is enabled, the error\n                # just gets swallowed by the logging module), and even if it were it would\n                # have the wrong stack trace. Just remember this fact and report it in\n                # __exit__ instead.\n                self.deprecated_level_matched += 1\n            if self.level is not None and record.levelno != self.level:\n                app_log.warning(\n                    \"Got expected log message %r at unexpected level (%s vs %s)\"\n                    % (message, logging.getLevelName(self.level), record.levelname)\n                )\n                return True\n            self.matched += 1\n            return False\n        return True\n\n    def __enter__(self) -> \"ExpectLog\":\n        if self.level is not None and self.level < self.logger.getEffectiveLevel():\n            self.orig_level = self.logger.level\n            self.logger.setLevel(self.level)\n        self.logger.addFilter(self)\n        return self\n\n    def __exit__(\n        self,\n        typ: \"Optional[Type[BaseException]]\",\n        value: Optional[BaseException],\n        tb: Optional[TracebackType],\n    ) -> None:\n        if self.orig_level is not None:\n            self.logger.setLevel(self.orig_level)\n        self.logger.removeFilter(self)\n        if not typ and self.required and not self.matched:\n            raise Exception(\"did not get expected log message\")\n        if (\n            not typ\n            and self.required\n            and (self.deprecated_level_matched >= self.matched)\n        ):\n            warnings.warn(\n                \"ExpectLog matched at INFO or below without level argument\",\n                DeprecationWarning,\n            )\n\n\n# From https://nedbatchelder.com/blog/201508/using_context_managers_in_test_setup.html\ndef setup_with_context_manager(testcase: unittest.TestCase, cm: Any) -> Any:\n    \"\"\"Use a contextmanager to setUp a test case.\"\"\"\n    val = cm.__enter__()\n    testcase.addCleanup(cm.__exit__, None, None, None)\n    return val\n\n\ndef main(**kwargs: Any) -> None:\n    \"\"\"A simple test runner.\n\n    This test runner is essentially equivalent to `unittest.main` from\n    the standard library, but adds support for Tornado-style option\n    parsing and log formatting. It is *not* necessary to use this\n    `main` function to run tests using `AsyncTestCase`; these tests\n    are self-contained and can run with any test runner.\n\n    The easiest way to run a test is via the command line::\n\n        python -m tornado.testing tornado.test.web_test\n\n    See the standard library ``unittest`` module for ways in which\n    tests can be specified.\n\n    Projects with many tests may wish to define a test script like\n    ``tornado/test/runtests.py``.  This script should define a method\n    ``all()`` which returns a test suite and then call\n    `tornado.testing.main()`.  Note that even when a test script is\n    used, the ``all()`` test suite may be overridden by naming a\n    single test on the command line::\n\n        # Runs all tests\n        python -m tornado.test.runtests\n        # Runs one test\n        python -m tornado.test.runtests tornado.test.web_test\n\n    Additional keyword arguments passed through to ``unittest.main()``.\n    For example, use ``tornado.testing.main(verbosity=2)``\n    to show many test details as they are run.\n    See http://docs.python.org/library/unittest.html#unittest.main\n    for full argument list.\n\n    .. versionchanged:: 5.0\n\n       This function produces no output of its own; only that produced\n       by the `unittest` module (previously it would add a PASS or FAIL\n       log message).\n    \"\"\"\n    from tornado.options import define, options, parse_command_line\n\n    define(\n        \"exception_on_interrupt\",\n        type=bool,\n        default=True,\n        help=(\n            \"If true (default), ctrl-c raises a KeyboardInterrupt \"\n            \"exception.  This prints a stack trace but cannot interrupt \"\n            \"certain operations.  If false, the process is more reliably \"\n            \"killed, but does not print a stack trace.\"\n        ),\n    )\n\n    # support the same options as unittest's command-line interface\n    define(\"verbose\", type=bool)\n    define(\"quiet\", type=bool)\n    define(\"failfast\", type=bool)\n    define(\"catch\", type=bool)\n    define(\"buffer\", type=bool)\n\n    argv = [sys.argv[0]] + parse_command_line(sys.argv)\n\n    if not options.exception_on_interrupt:\n        signal.signal(signal.SIGINT, signal.SIG_DFL)\n\n    if options.verbose is not None:\n        kwargs[\"verbosity\"] = 2\n    if options.quiet is not None:\n        kwargs[\"verbosity\"] = 0\n    if options.failfast is not None:\n        kwargs[\"failfast\"] = True\n    if options.catch is not None:\n        kwargs[\"catchbreak\"] = True\n    if options.buffer is not None:\n        kwargs[\"buffer\"] = True\n\n    if __name__ == \"__main__\" and len(argv) == 1:\n        print(\"No tests specified\", file=sys.stderr)\n        sys.exit(1)\n    # In order to be able to run tests by their fully-qualified name\n    # on the command line without importing all tests here,\n    # module must be set to None.  Python 3.2's unittest.main ignores\n    # defaultTest if no module is given (it tries to do its own\n    # test discovery, which is incompatible with auto2to3), so don't\n    # set module if we're not asking for a specific test.\n    if len(argv) > 1:\n        unittest.main(module=None, argv=argv, **kwargs)  # type: ignore\n    else:\n        unittest.main(defaultTest=\"all\", argv=argv, **kwargs)\n\n\nif __name__ == \"__main__\":\n    main()\n",
            "new_file_content": "\"\"\"Support classes for automated testing.\n\n* `AsyncTestCase` and `AsyncHTTPTestCase`:  Subclasses of unittest.TestCase\n  with additional support for testing asynchronous (`.IOLoop`-based) code.\n\n* `ExpectLog`: Make test logs less spammy.\n\n* `main()`: A simple test runner (wrapper around unittest.main()) with support\n  for the tornado.autoreload module to rerun the tests when code changes.\n\"\"\"\n\nimport asyncio\nfrom collections.abc import Generator\nimport functools\nimport inspect\nimport logging\nimport os\nimport re\nimport signal\nimport socket\nimport sys\nimport unittest\nimport warnings\n\nfrom tornado import gen\nfrom tornado.httpclient import AsyncHTTPClient, HTTPResponse\nfrom tornado.httpserver import HTTPServer\nfrom tornado.ioloop import IOLoop, TimeoutError\nfrom tornado import netutil\nfrom tornado.platform.asyncio import AsyncIOMainLoop\nfrom tornado.process import Subprocess\nfrom tornado.log import app_log\nfrom tornado.util import raise_exc_info, basestring_type\nfrom tornado.web import Application\n\nimport typing\nfrom typing import Tuple, Any, Callable, Type, Dict, Union, Optional, Coroutine\nfrom types import TracebackType\n\nif typing.TYPE_CHECKING:\n    _ExcInfoTuple = Tuple[\n        Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]\n    ]\n\n\n_NON_OWNED_IOLOOPS = AsyncIOMainLoop\n\n\ndef bind_unused_port(\n    reuse_port: bool = False, address: str = \"127.0.0.1\"\n) -> Tuple[socket.socket, int]:\n    \"\"\"Binds a server socket to an available port on localhost.\n\n    Returns a tuple (socket, port).\n\n    .. versionchanged:: 4.4\n       Always binds to ``127.0.0.1`` without resolving the name\n       ``localhost``.\n\n    .. versionchanged:: 6.2\n       Added optional ``address`` argument to\n       override the default \"127.0.0.1\".\n    \"\"\"\n    sock = netutil.bind_sockets(\n        0, address, family=socket.AF_INET, reuse_port=reuse_port\n    )[0]\n    port = sock.getsockname()[1]\n    return sock, port\n\n\ndef get_async_test_timeout() -> float:\n    \"\"\"Get the global timeout setting for async tests.\n\n    Returns a float, the timeout in seconds.\n\n    .. versionadded:: 3.1\n    \"\"\"\n    env = os.environ.get(\"ASYNC_TEST_TIMEOUT\")\n    if env is not None:\n        try:\n            return float(env)\n        except ValueError:\n            pass\n    return 5\n\n\nclass _TestMethodWrapper(object):\n    \"\"\"Wraps a test method to raise an error if it returns a value.\n\n    This is mainly used to detect undecorated generators (if a test\n    method yields it must use a decorator to consume the generator),\n    but will also detect other kinds of return values (these are not\n    necessarily errors, but we alert anyway since there is no good\n    reason to return a value from a test).\n    \"\"\"\n\n    def __init__(self, orig_method: Callable) -> None:\n        self.orig_method = orig_method\n        self.__wrapped__ = orig_method\n\n    def __call__(self, *args: Any, **kwargs: Any) -> None:\n        result = self.orig_method(*args, **kwargs)\n        if isinstance(result, Generator) or inspect.iscoroutine(result):\n            raise TypeError(\n                \"Generator and coroutine test methods should be\"\n                \" decorated with tornado.testing.gen_test\"\n            )\n        elif result is not None:\n            raise ValueError(\"Return value from test method ignored: %r\" % result)\n\n    def __getattr__(self, name: str) -> Any:\n        \"\"\"Proxy all unknown attributes to the original method.\n\n        This is important for some of the decorators in the `unittest`\n        module, such as `unittest.skipIf`.\n        \"\"\"\n        return getattr(self.orig_method, name)\n\n\nclass AsyncTestCase(unittest.TestCase):\n    \"\"\"`~unittest.TestCase` subclass for testing `.IOLoop`-based\n    asynchronous code.\n\n    The unittest framework is synchronous, so the test must be\n    complete by the time the test method returns. This means that\n    asynchronous code cannot be used in quite the same way as usual\n    and must be adapted to fit. To write your tests with coroutines,\n    decorate your test methods with `tornado.testing.gen_test` instead\n    of `tornado.gen.coroutine`.\n\n    This class also provides the (deprecated) `stop()` and `wait()`\n    methods for a more manual style of testing. The test method itself\n    must call ``self.wait()``, and asynchronous callbacks should call\n    ``self.stop()`` to signal completion.\n\n    By default, a new `.IOLoop` is constructed for each test and is available\n    as ``self.io_loop``.  If the code being tested requires a\n    global `.IOLoop`, subclasses should override `get_new_ioloop` to return it.\n\n    The `.IOLoop`'s ``start`` and ``stop`` methods should not be\n    called directly.  Instead, use `self.stop <stop>` and `self.wait\n    <wait>`.  Arguments passed to ``self.stop`` are returned from\n    ``self.wait``.  It is possible to have multiple ``wait``/``stop``\n    cycles in the same test.\n\n    Example::\n\n        # This test uses coroutine style.\n        class MyTestCase(AsyncTestCase):\n            @tornado.testing.gen_test\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                response = yield client.fetch(\"http://www.tornadoweb.org\")\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n\n        # This test uses argument passing between self.stop and self.wait.\n        class MyTestCase2(AsyncTestCase):\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                client.fetch(\"http://www.tornadoweb.org/\", self.stop)\n                response = self.wait()\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n    \"\"\"\n\n    def __init__(self, methodName: str = \"runTest\") -> None:\n        super().__init__(methodName)\n        self.__stopped = False\n        self.__running = False\n        self.__failure = None  # type: Optional[_ExcInfoTuple]\n        self.__stop_args = None  # type: Any\n        self.__timeout = None  # type: Optional[object]\n\n        # It's easy to forget the @gen_test decorator, but if you do\n        # the test will silently be ignored because nothing will consume\n        # the generator.  Replace the test method with a wrapper that will\n        # make sure it's not an undecorated generator.\n        setattr(self, methodName, _TestMethodWrapper(getattr(self, methodName)))\n\n        # Not used in this class itself, but used by @gen_test\n        self._test_generator = None  # type: Optional[Union[Generator, Coroutine]]\n\n    def setUp(self) -> None:\n        super().setUp()\n        self.io_loop = self.get_new_ioloop()\n        self.io_loop.make_current()\n\n    def tearDown(self) -> None:\n        # Native coroutines tend to produce warnings if they're not\n        # allowed to run to completion. It's difficult to ensure that\n        # this always happens in tests, so cancel any tasks that are\n        # still pending by the time we get here.\n        asyncio_loop = self.io_loop.asyncio_loop  # type: ignore\n        if hasattr(asyncio, \"all_tasks\"):  # py37\n            tasks = asyncio.all_tasks(asyncio_loop)  # type: ignore\n        else:\n            tasks = asyncio.Task.all_tasks(asyncio_loop)\n        # Tasks that are done may still appear here and may contain\n        # non-cancellation exceptions, so filter them out.\n        tasks = [t for t in tasks if not t.done()]  # type: ignore\n        for t in tasks:\n            t.cancel()\n        # Allow the tasks to run and finalize themselves (which means\n        # raising a CancelledError inside the coroutine). This may\n        # just transform the \"task was destroyed but it is pending\"\n        # warning into a \"uncaught CancelledError\" warning, but\n        # catching CancelledErrors in coroutines that may leak is\n        # simpler than ensuring that no coroutines leak.\n        if tasks:\n            done, pending = self.io_loop.run_sync(lambda: asyncio.wait(tasks))\n            assert not pending\n            # If any task failed with anything but a CancelledError, raise it.\n            for f in done:\n                try:\n                    f.result()\n                except asyncio.CancelledError:\n                    pass\n\n        # Clean up Subprocess, so it can be used again with a new ioloop.\n        Subprocess.uninitialize()\n        self.io_loop.clear_current()\n        if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):\n            # Try to clean up any file descriptors left open in the ioloop.\n            # This avoids leaks, especially when tests are run repeatedly\n            # in the same process with autoreload (because curl does not\n            # set FD_CLOEXEC on its file descriptors)\n            self.io_loop.close(all_fds=True)\n        super().tearDown()\n        # In case an exception escaped or the StackContext caught an exception\n        # when there wasn't a wait() to re-raise it, do so here.\n        # This is our last chance to raise an exception in a way that the\n        # unittest machinery understands.\n        self.__rethrow()\n\n    def get_new_ioloop(self) -> IOLoop:\n        \"\"\"Returns the `.IOLoop` to use for this test.\n\n        By default, a new `.IOLoop` is created for each test.\n        Subclasses may override this method to return\n        `.IOLoop.current()` if it is not appropriate to use a new\n        `.IOLoop` in each tests (for example, if there are global\n        singletons using the default `.IOLoop`) or if a per-test event\n        loop is being provided by another system (such as\n        ``pytest-asyncio``).\n        \"\"\"\n        return IOLoop(make_current=False)\n\n    def _handle_exception(\n        self, typ: Type[Exception], value: Exception, tb: TracebackType\n    ) -> bool:\n        if self.__failure is None:\n            self.__failure = (typ, value, tb)\n        else:\n            app_log.error(\n                \"multiple unhandled exceptions in test\", exc_info=(typ, value, tb)\n            )\n        self.stop()\n        return True\n\n    def __rethrow(self) -> None:\n        if self.__failure is not None:\n            failure = self.__failure\n            self.__failure = None\n            raise_exc_info(failure)\n\n    def run(\n        self, result: Optional[unittest.TestResult] = None\n    ) -> Optional[unittest.TestResult]:\n        ret = super().run(result)\n        # As a last resort, if an exception escaped super.run() and wasn't\n        # re-raised in tearDown, raise it here.  This will cause the\n        # unittest run to fail messily, but that's better than silently\n        # ignoring an error.\n        self.__rethrow()\n        return ret\n\n    def stop(self, _arg: Any = None, **kwargs: Any) -> None:\n        \"\"\"Stops the `.IOLoop`, causing one pending (or future) call to `wait()`\n        to return.\n\n        Keyword arguments or a single positional argument passed to `stop()` are\n        saved and will be returned by `wait()`.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        assert _arg is None or not kwargs\n        self.__stop_args = kwargs or _arg\n        if self.__running:\n            self.io_loop.stop()\n            self.__running = False\n        self.__stopped = True\n\n    def wait(\n        self,\n        condition: Optional[Callable[..., bool]] = None,\n        timeout: Optional[float] = None,\n    ) -> Any:\n        \"\"\"Runs the `.IOLoop` until stop is called or timeout has passed.\n\n        In the event of a timeout, an exception will be thrown. The\n        default timeout is 5 seconds; it may be overridden with a\n        ``timeout`` keyword argument or globally with the\n        ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        If ``condition`` is not ``None``, the `.IOLoop` will be restarted\n        after `stop()` until ``condition()`` returns ``True``.\n\n        .. versionchanged:: 3.1\n           Added the ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        if timeout is None:\n            timeout = get_async_test_timeout()\n\n        if not self.__stopped:\n            if timeout:\n\n                def timeout_func() -> None:\n                    try:\n                        raise self.failureException(\n                            \"Async operation timed out after %s seconds\" % timeout\n                        )\n                    except Exception:\n                        self.__failure = sys.exc_info()\n                    self.stop()\n\n                self.__timeout = self.io_loop.add_timeout(\n                    self.io_loop.time() + timeout, timeout_func\n                )\n            while True:\n                self.__running = True\n                self.io_loop.start()\n                if self.__failure is not None or condition is None or condition():\n                    break\n            if self.__timeout is not None:\n                self.io_loop.remove_timeout(self.__timeout)\n                self.__timeout = None\n        assert self.__stopped\n        self.__stopped = False\n        self.__rethrow()\n        result = self.__stop_args\n        self.__stop_args = None\n        return result\n\n\nclass AsyncHTTPTestCase(AsyncTestCase):\n    \"\"\"A test case that starts up an HTTP server.\n\n    Subclasses must override `get_app()`, which returns the\n    `tornado.web.Application` (or other `.HTTPServer` callback) to be tested.\n    Tests will typically use the provided ``self.http_client`` to fetch\n    URLs from this server.\n\n    Example, assuming the \"Hello, world\" example from the user guide is in\n    ``hello.py``::\n\n        import hello\n\n        class TestHelloApp(AsyncHTTPTestCase):\n            def get_app(self):\n                return hello.make_app()\n\n            def test_homepage(self):\n                response = self.fetch('/')\n                self.assertEqual(response.code, 200)\n                self.assertEqual(response.body, 'Hello, world')\n\n    That call to ``self.fetch()`` is equivalent to ::\n\n        self.http_client.fetch(self.get_url('/'), self.stop)\n        response = self.wait()\n\n    which illustrates how AsyncTestCase can turn an asynchronous operation,\n    like ``http_client.fetch()``, into a synchronous operation. If you need\n    to do other asynchronous operations in tests, you'll probably need to use\n    ``stop()`` and ``wait()`` yourself.\n    \"\"\"\n\n    def setUp(self) -> None:\n        super().setUp()\n        sock, port = bind_unused_port()\n        self.__port = port\n\n        self.http_client = self.get_http_client()\n        self._app = self.get_app()\n        self.http_server = self.get_http_server()\n        self.http_server.add_sockets([sock])\n\n    def get_http_client(self) -> AsyncHTTPClient:\n        return AsyncHTTPClient()\n\n    def get_http_server(self) -> HTTPServer:\n        return HTTPServer(self._app, **self.get_httpserver_options())\n\n    def get_app(self) -> Application:\n        \"\"\"Should be overridden by subclasses to return a\n        `tornado.web.Application` or other `.HTTPServer` callback.\n        \"\"\"\n        raise NotImplementedError()\n\n    def fetch(\n        self, path: str, raise_error: bool = False, **kwargs: Any\n    ) -> HTTPResponse:\n        \"\"\"Convenience method to synchronously fetch a URL.\n\n        The given path will be appended to the local server's host and\n        port.  Any additional keyword arguments will be passed directly to\n        `.AsyncHTTPClient.fetch` (and so could be used to pass\n        ``method=\"POST\"``, ``body=\"...\"``, etc).\n\n        If the path begins with http:// or https://, it will be treated as a\n        full URL and will be fetched as-is.\n\n        If ``raise_error`` is ``True``, a `tornado.httpclient.HTTPError` will\n        be raised if the response code is not 200. This is the same behavior\n        as the ``raise_error`` argument to `.AsyncHTTPClient.fetch`, but\n        the default is ``False`` here (it's ``True`` in `.AsyncHTTPClient`)\n        because tests often need to deal with non-200 response codes.\n\n        .. versionchanged:: 5.0\n           Added support for absolute URLs.\n\n        .. versionchanged:: 5.1\n\n           Added the ``raise_error`` argument.\n\n        .. deprecated:: 5.1\n\n           This method currently turns any exception into an\n           `.HTTPResponse` with status code 599. In Tornado 6.0,\n           errors other than `tornado.httpclient.HTTPError` will be\n           passed through, and ``raise_error=False`` will only\n           suppress errors that would be raised due to non-200\n           response codes.\n\n        \"\"\"\n        if path.lower().startswith((\"http://\", \"https://\")):\n            url = path\n        else:\n            url = self.get_url(path)\n        return self.io_loop.run_sync(\n            lambda: self.http_client.fetch(url, raise_error=raise_error, **kwargs),\n            timeout=get_async_test_timeout(),\n        )\n\n    def get_httpserver_options(self) -> Dict[str, Any]:\n        \"\"\"May be overridden by subclasses to return additional\n        keyword arguments for the server.\n        \"\"\"\n        return {}\n\n    def get_http_port(self) -> int:\n        \"\"\"Returns the port used by the server.\n\n        A new port is chosen for each test.\n        \"\"\"\n        return self.__port\n\n    def get_protocol(self) -> str:\n        return \"http\"\n\n    def get_url(self, path: str) -> str:\n        \"\"\"Returns an absolute url for the given path on the test server.\"\"\"\n        return \"%s://127.0.0.1:%s%s\" % (self.get_protocol(), self.get_http_port(), path)\n\n    def tearDown(self) -> None:\n        self.http_server.stop()\n        self.io_loop.run_sync(\n            self.http_server.close_all_connections, timeout=get_async_test_timeout()\n        )\n        self.http_client.close()\n        del self.http_server\n        del self._app\n        super().tearDown()\n\n\nclass AsyncHTTPSTestCase(AsyncHTTPTestCase):\n    \"\"\"A test case that starts an HTTPS server.\n\n    Interface is generally the same as `AsyncHTTPTestCase`.\n    \"\"\"\n\n    def get_http_client(self) -> AsyncHTTPClient:\n        return AsyncHTTPClient(force_instance=True, defaults=dict(validate_cert=False))\n\n    def get_httpserver_options(self) -> Dict[str, Any]:\n        return dict(ssl_options=self.get_ssl_options())\n\n    def get_ssl_options(self) -> Dict[str, Any]:\n        \"\"\"May be overridden by subclasses to select SSL options.\n\n        By default includes a self-signed testing certificate.\n        \"\"\"\n        return AsyncHTTPSTestCase.default_ssl_options()\n\n    @staticmethod\n    def default_ssl_options() -> Dict[str, Any]:\n        # Testing keys were generated with:\n        # openssl req -new -keyout tornado/test/test.key \\\n        #                     -out tornado/test/test.crt -nodes -days 3650 -x509\n        module_dir = os.path.dirname(__file__)\n        return dict(\n            certfile=os.path.join(module_dir, \"test\", \"test.crt\"),\n            keyfile=os.path.join(module_dir, \"test\", \"test.key\"),\n        )\n\n    def get_protocol(self) -> str:\n        return \"https\"\n\n\n@typing.overload\ndef gen_test(\n    *, timeout: Optional[float] = None\n) -> Callable[[Callable[..., Union[Generator, \"Coroutine\"]]], Callable[..., None]]:\n    pass\n\n\n@typing.overload  # noqa: F811\ndef gen_test(func: Callable[..., Union[Generator, \"Coroutine\"]]) -> Callable[..., None]:\n    pass\n\n\ndef gen_test(  # noqa: F811\n    func: Optional[Callable[..., Union[Generator, \"Coroutine\"]]] = None,\n    timeout: Optional[float] = None,\n) -> Union[\n    Callable[..., None],\n    Callable[[Callable[..., Union[Generator, \"Coroutine\"]]], Callable[..., None]],\n]:\n    \"\"\"Testing equivalent of ``@gen.coroutine``, to be applied to test methods.\n\n    ``@gen.coroutine`` cannot be used on tests because the `.IOLoop` is not\n    already running.  ``@gen_test`` should be applied to test methods\n    on subclasses of `AsyncTestCase`.\n\n    Example::\n\n        class MyTest(AsyncHTTPTestCase):\n            @gen_test\n            def test_something(self):\n                response = yield self.http_client.fetch(self.get_url('/'))\n\n    By default, ``@gen_test`` times out after 5 seconds. The timeout may be\n    overridden globally with the ``ASYNC_TEST_TIMEOUT`` environment variable,\n    or for each test with the ``timeout`` keyword argument::\n\n        class MyTest(AsyncHTTPTestCase):\n            @gen_test(timeout=10)\n            def test_something_slow(self):\n                response = yield self.http_client.fetch(self.get_url('/'))\n\n    Note that ``@gen_test`` is incompatible with `AsyncTestCase.stop`,\n    `AsyncTestCase.wait`, and `AsyncHTTPTestCase.fetch`. Use ``yield\n    self.http_client.fetch(self.get_url())`` as shown above instead.\n\n    .. versionadded:: 3.1\n       The ``timeout`` argument and ``ASYNC_TEST_TIMEOUT`` environment\n       variable.\n\n    .. versionchanged:: 4.0\n       The wrapper now passes along ``*args, **kwargs`` so it can be used\n       on functions with arguments.\n\n    \"\"\"\n    if timeout is None:\n        timeout = get_async_test_timeout()\n\n    def wrap(f: Callable[..., Union[Generator, \"Coroutine\"]]) -> Callable[..., None]:\n        # Stack up several decorators to allow us to access the generator\n        # object itself.  In the innermost wrapper, we capture the generator\n        # and save it in an attribute of self.  Next, we run the wrapped\n        # function through @gen.coroutine.  Finally, the coroutine is\n        # wrapped again to make it synchronous with run_sync.\n        #\n        # This is a good case study arguing for either some sort of\n        # extensibility in the gen decorators or cancellation support.\n        @functools.wraps(f)\n        def pre_coroutine(self, *args, **kwargs):\n            # type: (AsyncTestCase, *Any, **Any) -> Union[Generator, Coroutine]\n            # Type comments used to avoid pypy3 bug.\n            result = f(self, *args, **kwargs)\n            if isinstance(result, Generator) or inspect.iscoroutine(result):\n                self._test_generator = result\n            else:\n                self._test_generator = None\n            return result\n\n        if inspect.iscoroutinefunction(f):\n            coro = pre_coroutine\n        else:\n            coro = gen.coroutine(pre_coroutine)  # type: ignore[assignment]\n\n        @functools.wraps(coro)\n        def post_coroutine(self, *args, **kwargs):\n            # type: (AsyncTestCase, *Any, **Any) -> None\n            try:\n                return self.io_loop.run_sync(\n                    functools.partial(coro, self, *args, **kwargs), timeout=timeout\n                )\n            except TimeoutError as e:\n                # run_sync raises an error with an unhelpful traceback.\n                # If the underlying generator is still running, we can throw the\n                # exception back into it so the stack trace is replaced by the\n                # point where the test is stopped. The only reason the generator\n                # would not be running would be if it were cancelled, which means\n                # a native coroutine, so we can rely on the cr_running attribute.\n                if self._test_generator is not None and getattr(\n                    self._test_generator, \"cr_running\", True\n                ):\n                    self._test_generator.throw(e)\n                    # In case the test contains an overly broad except\n                    # clause, we may get back here.\n                # Coroutine was stopped or didn't raise a useful stack trace,\n                # so re-raise the original exception which is better than nothing.\n                raise\n\n        return post_coroutine\n\n    if func is not None:\n        # Used like:\n        #     @gen_test\n        #     def f(self):\n        #         pass\n        return wrap(func)\n    else:\n        # Used like @gen_test(timeout=10)\n        return wrap\n\n\n# Without this attribute, nosetests will try to run gen_test as a test\n# anywhere it is imported.\ngen_test.__test__ = False  # type: ignore\n\n\nclass ExpectLog(logging.Filter):\n    \"\"\"Context manager to capture and suppress expected log output.\n\n    Useful to make tests of error conditions less noisy, while still\n    leaving unexpected log entries visible.  *Not thread safe.*\n\n    The attribute ``logged_stack`` is set to ``True`` if any exception\n    stack trace was logged.\n\n    Usage::\n\n        with ExpectLog('tornado.application', \"Uncaught exception\"):\n            error_response = self.fetch(\"/some_page\")\n\n    .. versionchanged:: 4.3\n       Added the ``logged_stack`` attribute.\n    \"\"\"\n\n    def __init__(\n        self,\n        logger: Union[logging.Logger, basestring_type],\n        regex: str,\n        required: bool = True,\n        level: Optional[int] = None,\n    ) -> None:\n        \"\"\"Constructs an ExpectLog context manager.\n\n        :param logger: Logger object (or name of logger) to watch.  Pass an\n            empty string to watch the root logger.\n        :param regex: Regular expression to match.  Any log entries on the\n            specified logger that match this regex will be suppressed.\n        :param required: If true, an exception will be raised if the end of the\n            ``with`` statement is reached without matching any log entries.\n        :param level: A constant from the ``logging`` module indicating the\n            expected log level. If this parameter is provided, only log messages\n            at this level will be considered to match. Additionally, the\n            supplied ``logger`` will have its level adjusted if necessary (for\n            the duration of the ``ExpectLog`` to enable the expected message.\n\n        .. versionchanged:: 6.1\n           Added the ``level`` parameter.\n\n        .. deprecated:: 6.3\n           In Tornado 7.0, only ``WARNING`` and higher logging levels will be\n           matched by default. To match ``INFO`` and lower levels, the ``level``\n           argument must be used. This is changing to minimize differences\n           between ``tornado.testing.main`` (which enables ``INFO`` logs by\n           default) and most other test runners (including those in IDEs)\n           which have ``INFO`` logs disabled by default.\n        \"\"\"\n        if isinstance(logger, basestring_type):\n            logger = logging.getLogger(logger)\n        self.logger = logger\n        self.regex = re.compile(regex)\n        self.required = required\n        # matched and deprecated_level_matched are a counter for the respective event.\n        self.matched = 0\n        self.deprecated_level_matched = 0\n        self.logged_stack = False\n        self.level = level\n        self.orig_level = None  # type: Optional[int]\n\n    def filter(self, record: logging.LogRecord) -> bool:\n        if record.exc_info:\n            self.logged_stack = True\n        message = record.getMessage()\n        if self.regex.match(message):\n            if self.level is None and record.levelno < logging.WARNING:\n                # We're inside the logging machinery here so generating a DeprecationWarning\n                # here won't be reported cleanly (if warnings-as-errors is enabled, the error\n                # just gets swallowed by the logging module), and even if it were it would\n                # have the wrong stack trace. Just remember this fact and report it in\n                # __exit__ instead.\n                self.deprecated_level_matched += 1\n            if self.level is not None and record.levelno != self.level:\n                app_log.warning(\n                    \"Got expected log message %r at unexpected level (%s vs %s)\"\n                    % (message, logging.getLevelName(self.level), record.levelname)\n                )\n                return True\n            self.matched += 1\n            return False\n        return True\n\n    def __enter__(self) -> \"ExpectLog\":\n        if self.level is not None and self.level < self.logger.getEffectiveLevel():\n            self.orig_level = self.logger.level\n            self.logger.setLevel(self.level)\n        self.logger.addFilter(self)\n        return self\n\n    def __exit__(\n        self,\n        typ: \"Optional[Type[BaseException]]\",\n        value: Optional[BaseException],\n        tb: Optional[TracebackType],\n    ) -> None:\n        if self.orig_level is not None:\n            self.logger.setLevel(self.orig_level)\n        self.logger.removeFilter(self)\n        if not typ and self.required and not self.matched:\n            raise Exception(\"did not get expected log message\")\n        if (\n            not typ\n            and self.required\n            and (self.deprecated_level_matched >= self.matched)\n        ):\n            warnings.warn(\n                \"ExpectLog matched at INFO or below without level argument\",\n                DeprecationWarning,\n            )\n\n\n# From https://nedbatchelder.com/blog/201508/using_context_managers_in_test_setup.html\ndef setup_with_context_manager(testcase: unittest.TestCase, cm: Any) -> Any:\n    \"\"\"Use a contextmanager to setUp a test case.\"\"\"\n    val = cm.__enter__()\n    testcase.addCleanup(cm.__exit__, None, None, None)\n    return val\n\n\ndef main(**kwargs: Any) -> None:\n    \"\"\"A simple test runner.\n\n    This test runner is essentially equivalent to `unittest.main` from\n    the standard library, but adds support for Tornado-style option\n    parsing and log formatting. It is *not* necessary to use this\n    `main` function to run tests using `AsyncTestCase`; these tests\n    are self-contained and can run with any test runner.\n\n    The easiest way to run a test is via the command line::\n\n        python -m tornado.testing tornado.test.web_test\n\n    See the standard library ``unittest`` module for ways in which\n    tests can be specified.\n\n    Projects with many tests may wish to define a test script like\n    ``tornado/test/runtests.py``.  This script should define a method\n    ``all()`` which returns a test suite and then call\n    `tornado.testing.main()`.  Note that even when a test script is\n    used, the ``all()`` test suite may be overridden by naming a\n    single test on the command line::\n\n        # Runs all tests\n        python -m tornado.test.runtests\n        # Runs one test\n        python -m tornado.test.runtests tornado.test.web_test\n\n    Additional keyword arguments passed through to ``unittest.main()``.\n    For example, use ``tornado.testing.main(verbosity=2)``\n    to show many test details as they are run.\n    See http://docs.python.org/library/unittest.html#unittest.main\n    for full argument list.\n\n    .. versionchanged:: 5.0\n\n       This function produces no output of its own; only that produced\n       by the `unittest` module (previously it would add a PASS or FAIL\n       log message).\n    \"\"\"\n    from tornado.options import define, options, parse_command_line\n\n    define(\n        \"exception_on_interrupt\",\n        type=bool,\n        default=True,\n        help=(\n            \"If true (default), ctrl-c raises a KeyboardInterrupt \"\n            \"exception.  This prints a stack trace but cannot interrupt \"\n            \"certain operations.  If false, the process is more reliably \"\n            \"killed, but does not print a stack trace.\"\n        ),\n    )\n\n    # support the same options as unittest's command-line interface\n    define(\"verbose\", type=bool)\n    define(\"quiet\", type=bool)\n    define(\"failfast\", type=bool)\n    define(\"catch\", type=bool)\n    define(\"buffer\", type=bool)\n\n    argv = [sys.argv[0]] + parse_command_line(sys.argv)\n\n    if not options.exception_on_interrupt:\n        signal.signal(signal.SIGINT, signal.SIG_DFL)\n\n    if options.verbose is not None:\n        kwargs[\"verbosity\"] = 2\n    if options.quiet is not None:\n        kwargs[\"verbosity\"] = 0\n    if options.failfast is not None:\n        kwargs[\"failfast\"] = True\n    if options.catch is not None:\n        kwargs[\"catchbreak\"] = True\n    if options.buffer is not None:\n        kwargs[\"buffer\"] = True\n\n    if __name__ == \"__main__\" and len(argv) == 1:\n        print(\"No tests specified\", file=sys.stderr)\n        sys.exit(1)\n    # In order to be able to run tests by their fully-qualified name\n    # on the command line without importing all tests here,\n    # module must be set to None.  Python 3.2's unittest.main ignores\n    # defaultTest if no module is given (it tries to do its own\n    # test discovery, which is incompatible with auto2to3), so don't\n    # set module if we're not asking for a specific test.\n    if len(argv) > 1:\n        unittest.main(module=None, argv=argv, **kwargs)  # type: ignore\n    else:\n        unittest.main(defaultTest=\"all\", argv=argv, **kwargs)\n\n\nif __name__ == \"__main__\":\n    main()\n",
            "header": {
                "file": {
                    "path": "tornado/testing.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "0232b219",
                "new_commit_hash": "508b1273",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/tornado/testing.py"
            },
            "plus_file": {
                "path": "b/tornado/testing.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 182,
                            "length": 49
                        },
                        "new_range": {
                            "start": 182,
                            "length": 9
                        },
                        "section": "class AsyncTestCase(unittest.TestCase):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        self._test_generator = None  # type: Optional[Union[Generator, Coroutine]]",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    def setUp(self) -> None:",
                                "type": "context"
                            },
                            {
                                "content": "        setup_with_context_manager(self, warnings.catch_warnings())",
                                "type": "deleted"
                            },
                            {
                                "content": "        warnings.filterwarnings(",
                                "type": "deleted"
                            },
                            {
                                "content": "            \"ignore\",",
                                "type": "deleted"
                            },
                            {
                                "content": "            message=\"There is no current event loop\",",
                                "type": "deleted"
                            },
                            {
                                "content": "            category=DeprecationWarning,",
                                "type": "deleted"
                            },
                            {
                                "content": "            module=r\"tornado\\..*\",",
                                "type": "deleted"
                            },
                            {
                                "content": "        )",
                                "type": "deleted"
                            },
                            {
                                "content": "        super().setUp()",
                                "type": "context"
                            },
                            {
                                "content": "        # NOTE: this code attempts to navigate deprecation warnings introduced",
                                "type": "deleted"
                            },
                            {
                                "content": "        # in Python 3.10. The idea of an implicit current event loop is",
                                "type": "deleted"
                            },
                            {
                                "content": "        # deprecated in that version, with the intention that tests like this",
                                "type": "deleted"
                            },
                            {
                                "content": "        # explicitly create a new event loop and run on it. However, other",
                                "type": "deleted"
                            },
                            {
                                "content": "        # packages such as pytest-asyncio (as of version 0.16.0) still rely on",
                                "type": "deleted"
                            },
                            {
                                "content": "        # the implicit current event loop and we want to be compatible with them",
                                "type": "deleted"
                            },
                            {
                                "content": "        # (even when run on 3.10, but not, of course, on the future version of",
                                "type": "deleted"
                            },
                            {
                                "content": "        # python that removes the get/set_event_loop methods completely).",
                                "type": "deleted"
                            },
                            {
                                "content": "        #",
                                "type": "deleted"
                            },
                            {
                                "content": "        # Deprecation warnings were introduced inconsistently:",
                                "type": "deleted"
                            },
                            {
                                "content": "        # asyncio.get_event_loop warns, but",
                                "type": "deleted"
                            },
                            {
                                "content": "        # asyncio.get_event_loop_policy().get_event_loop does not. Similarly,",
                                "type": "deleted"
                            },
                            {
                                "content": "        # none of the set_event_loop methods warn, although comments on",
                                "type": "deleted"
                            },
                            {
                                "content": "        # https://bugs.python.org/issue39529 indicate that they are also",
                                "type": "deleted"
                            },
                            {
                                "content": "        # intended for future removal.",
                                "type": "deleted"
                            },
                            {
                                "content": "        #",
                                "type": "deleted"
                            },
                            {
                                "content": "        # Therefore, we first attempt to access the event loop with the",
                                "type": "deleted"
                            },
                            {
                                "content": "        # (non-warning) policy method, and if it fails, fall back to creating a",
                                "type": "deleted"
                            },
                            {
                                "content": "        # new event loop. We do not have effective test coverage of the",
                                "type": "deleted"
                            },
                            {
                                "content": "        # new event loop case; this will have to be watched when/if",
                                "type": "deleted"
                            },
                            {
                                "content": "        # get_event_loop is actually removed.",
                                "type": "deleted"
                            },
                            {
                                "content": "        self.should_close_asyncio_loop = False",
                                "type": "deleted"
                            },
                            {
                                "content": "        try:",
                                "type": "deleted"
                            },
                            {
                                "content": "            self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()",
                                "type": "deleted"
                            },
                            {
                                "content": "        except Exception:",
                                "type": "deleted"
                            },
                            {
                                "content": "            self.asyncio_loop = asyncio.new_event_loop()",
                                "type": "deleted"
                            },
                            {
                                "content": "            self.should_close_asyncio_loop = True",
                                "type": "deleted"
                            },
                            {
                                "content": "",
                                "type": "deleted"
                            },
                            {
                                "content": "        async def get_loop() -> IOLoop:",
                                "type": "deleted"
                            },
                            {
                                "content": "            return self.get_new_ioloop()",
                                "type": "deleted"
                            },
                            {
                                "content": "",
                                "type": "deleted"
                            },
                            {
                                "content": "        self.io_loop = self.asyncio_loop.run_until_complete(get_loop())",
                                "type": "deleted"
                            },
                            {
                                "content": "        with warnings.catch_warnings():",
                                "type": "deleted"
                            },
                            {
                                "content": "            warnings.simplefilter(\"ignore\", DeprecationWarning)",
                                "type": "deleted"
                            },
                            {
                                "content": "            self.io_loop.make_current()",
                                "type": "deleted"
                            },
                            {
                                "content": "        self.io_loop = self.get_new_ioloop()",
                                "type": "added"
                            },
                            {
                                "content": "        self.io_loop.make_current()",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    def tearDown(self) -> None:",
                                "type": "context"
                            },
                            {
                                "content": "        # Native coroutines tend to produce warnings if they're not",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/testing.py",
                            "type": "method",
                            "name": "AsyncTestCase.setUp",
                            "content": "class AsyncTestCase(unittest.TestCase):\n    \"\"\"`~unittest.TestCase` subclass for testing `.IOLoop`-based\n    asynchronous code.\n\n    The unittest framework is synchronous, so the test must be\n    complete by the time the test method returns. This means that\n    asynchronous code cannot be used in quite the same way as usual\n    and must be adapted to fit. To write your tests with coroutines,\n    decorate your test methods with `tornado.testing.gen_test` instead\n    of `tornado.gen.coroutine`.\n\n    This class also provides the (deprecated) `stop()` and `wait()`\n    methods for a more manual style of testing. The test method itself\n    must call ``self.wait()``, and asynchronous callbacks should call\n    ``self.stop()`` to signal completion.\n\n    By default, a new `.IOLoop` is constructed for each test and is available\n    as ``self.io_loop``.  If the code being tested requires a\n    global `.IOLoop`, subclasses should override `get_new_ioloop` to return it.\n\n    The `.IOLoop`'s ``start`` and ``stop`` methods should not be\n    called directly.  Instead, use `self.stop <stop>` and `self.wait\n    <wait>`.  Arguments passed to ``self.stop`` are returned from\n    ``self.wait``.  It is possible to have multiple ``wait``/``stop``\n    cycles in the same test.\n\n    Example::\n\n        # This test uses coroutine style.\n        class MyTestCase(AsyncTestCase):\n            @tornado.testing.gen_test\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                response = yield client.fetch(\"http://www.tornadoweb.org\")\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n\n        # This test uses argument passing between self.stop and self.wait.\n        class MyTestCase2(AsyncTestCase):\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                client.fetch(\"http://www.tornadoweb.org/\", self.stop)\n                response = self.wait()\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n    \"\"\"\n\n    def __init__(self, methodName: str = \"runTest\") -> None:\n        super().__init__(methodName)\n        self.__stopped = False\n        self.__running = False\n        self.__failure = None  # type: Optional[_ExcInfoTuple]\n        self.__stop_args = None  # type: Any\n        self.__timeout = None  # type: Optional[object]\n\n        # It's easy to forget the @gen_test decorator, but if you do\n        # the test will silently be ignored because nothing will consume\n        # the generator.  Replace the test method with a wrapper that will\n        # make sure it's not an undecorated generator.\n        setattr(self, methodName, _TestMethodWrapper(getattr(self, methodName)))\n\n        # Not used in this class itself, but used by @gen_test\n        self._test_generator = None  # type: Optional[Union[Generator, Coroutine]]\n\n    def setUp(self) -> None:\n        setup_with_context_manager(self, warnings.catch_warnings())\n        warnings.filterwarnings(\n            \"ignore\",\n            message=\"There is no current event loop\",\n            category=DeprecationWarning,\n            module=r\"tornado\\..*\",\n        )\n        super().setUp()\n        # NOTE: this code attempts to navigate deprecation warnings introduced\n        # in Python 3.10. The idea of an implicit current event loop is\n        # deprecated in that version, with the intention that tests like this\n        # explicitly create a new event loop and run on it. However, other\n        # packages such as pytest-asyncio (as of version 0.16.0) still rely on\n        # the implicit current event loop and we want to be compatible with them\n        # (even when run on 3.10, but not, of course, on the future version of\n        # python that removes the get/set_event_loop methods completely).\n        #\n        # Deprecation warnings were introduced inconsistently:\n        # asyncio.get_event_loop warns, but\n        # asyncio.get_event_loop_policy().get_event_loop does not. Similarly,\n        # none of the set_event_loop methods warn, although comments on\n        # https://bugs.python.org/issue39529 indicate that they are also\n        # intended for future removal.\n        #\n        # Therefore, we first attempt to access the event loop with the\n        # (non-warning) policy method, and if it fails, fall back to creating a\n        # new event loop. We do not have effective test coverage of the\n        # new event loop case; this will have to be watched when/if\n        # get_event_loop is actually removed.\n        self.should_close_asyncio_loop = False\n        try:\n            self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()\n        except Exception:\n            self.asyncio_loop = asyncio.new_event_loop()\n            self.should_close_asyncio_loop = True\n\n        async def get_loop() -> IOLoop:\n            return self.get_new_ioloop()\n\n        self.io_loop = self.asyncio_loop.run_until_complete(get_loop())\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.make_current()\n\n    def tearDown(self) -> None:\n        # Native coroutines tend to produce warnings if they're not\n        # allowed to run to completion. It's difficult to ensure that\n        # this always happens in tests, so cancel any tasks that are\n        # still pending by the time we get here.\n        asyncio_loop = self.io_loop.asyncio_loop  # type: ignore\n        if hasattr(asyncio, \"all_tasks\"):  # py37\n            tasks = asyncio.all_tasks(asyncio_loop)  # type: ignore\n        else:\n            tasks = asyncio.Task.all_tasks(asyncio_loop)\n        # Tasks that are done may still appear here and may contain\n        # non-cancellation exceptions, so filter them out.\n        tasks = [t for t in tasks if not t.done()]  # type: ignore\n        for t in tasks:\n            t.cancel()\n        # Allow the tasks to run and finalize themselves (which means\n        # raising a CancelledError inside the coroutine). This may\n        # just transform the \"task was destroyed but it is pending\"\n        # warning into a \"uncaught CancelledError\" warning, but\n        # catching CancelledErrors in coroutines that may leak is\n        # simpler than ensuring that no coroutines leak.\n        if tasks:\n            done, pending = self.io_loop.run_sync(lambda: asyncio.wait(tasks))\n            assert not pending\n            # If any task failed with anything but a CancelledError, raise it.\n            for f in done:\n                try:\n                    f.result()\n                except asyncio.CancelledError:\n                    pass\n\n        # Clean up Subprocess, so it can be used again with a new ioloop.\n        Subprocess.uninitialize()\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.clear_current()\n        if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):\n            # Try to clean up any file descriptors left open in the ioloop.\n            # This avoids leaks, especially when tests are run repeatedly\n            # in the same process with autoreload (because curl does not\n            # set FD_CLOEXEC on its file descriptors)\n            self.io_loop.close(all_fds=True)\n        if self.should_close_asyncio_loop:\n            self.asyncio_loop.close()\n        super().tearDown()\n        # In case an exception escaped or the StackContext caught an exception\n        # when there wasn't a wait() to re-raise it, do so here.\n        # This is our last chance to raise an exception in a way that the\n        # unittest machinery understands.\n        self.__rethrow()\n\n    def get_new_ioloop(self) -> IOLoop:\n        \"\"\"Returns the `.IOLoop` to use for this test.\n\n        By default, a new `.IOLoop` is created for each test.\n        Subclasses may override this method to return\n        `.IOLoop.current()` if it is not appropriate to use a new\n        `.IOLoop` in each tests (for example, if there are global\n        singletons using the default `.IOLoop`) or if a per-test event\n        loop is being provided by another system (such as\n        ``pytest-asyncio``).\n        \"\"\"\n        return IOLoop(make_current=False)\n\n    def _handle_exception(\n        self, typ: Type[Exception], value: Exception, tb: TracebackType\n    ) -> bool:\n        if self.__failure is None:\n            self.__failure = (typ, value, tb)\n        else:\n            app_log.error(\n                \"multiple unhandled exceptions in test\", exc_info=(typ, value, tb)\n            )\n        self.stop()\n        return True\n\n    def __rethrow(self) -> None:\n        if self.__failure is not None:\n            failure = self.__failure\n            self.__failure = None\n            raise_exc_info(failure)\n\n    def run(\n        self, result: Optional[unittest.TestResult] = None\n    ) -> Optional[unittest.TestResult]:\n        ret = super().run(result)\n        # As a last resort, if an exception escaped super.run() and wasn't\n        # re-raised in tearDown, raise it here.  This will cause the\n        # unittest run to fail messily, but that's better than silently\n        # ignoring an error.\n        self.__rethrow()\n        return ret\n\n    def stop(self, _arg: Any = None, **kwargs: Any) -> None:\n        \"\"\"Stops the `.IOLoop`, causing one pending (or future) call to `wait()`\n        to return.\n\n        Keyword arguments or a single positional argument passed to `stop()` are\n        saved and will be returned by `wait()`.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        assert _arg is None or not kwargs\n        self.__stop_args = kwargs or _arg\n        if self.__running:\n            self.io_loop.stop()\n            self.__running = False\n        self.__stopped = True\n\n    def wait(\n        self,\n        condition: Optional[Callable[..., bool]] = None,\n        timeout: Optional[float] = None,\n    ) -> Any:\n        \"\"\"Runs the `.IOLoop` until stop is called or timeout has passed.\n\n        In the event of a timeout, an exception will be thrown. The\n        default timeout is 5 seconds; it may be overridden with a\n        ``timeout`` keyword argument or globally with the\n        ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        If ``condition`` is not ``None``, the `.IOLoop` will be restarted\n        after `stop()` until ``condition()`` returns ``True``.\n\n        .. versionchanged:: 3.1\n           Added the ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        if timeout is None:\n            timeout = get_async_test_timeout()\n\n        if not self.__stopped:\n            if timeout:\n\n                def timeout_func() -> None:\n                    try:\n                        raise self.failureException(\n                            \"Async operation timed out after %s seconds\" % timeout\n                        )\n                    except Exception:\n                        self.__failure = sys.exc_info()\n                    self.stop()\n\n                self.__timeout = self.io_loop.add_timeout(\n                    self.io_loop.time() + timeout, timeout_func\n                )\n            while True:\n                self.__running = True\n                self.io_loop.start()\n                if self.__failure is not None or condition is None or condition():\n                    break\n            if self.__timeout is not None:\n                self.io_loop.remove_timeout(self.__timeout)\n                self.__timeout = None\n        assert self.__stopped\n        self.__stopped = False\n        self.__rethrow()\n        result = self.__stop_args\n        self.__stop_args = None\n        return result",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 184,
                            "end_lineno": 227,
                            "parent": {
                                "file_name": "tornado/testing.py",
                                "type": "class",
                                "name": "AsyncTestCase",
                                "content": "class AsyncTestCase(unittest.TestCase):\n    \"\"\"`~unittest.TestCase` subclass for testing `.IOLoop`-based\n    asynchronous code.\n\n    The unittest framework is synchronous, so the test must be\n    complete by the time the test method returns. This means that\n    asynchronous code cannot be used in quite the same way as usual\n    and must be adapted to fit. To write your tests with coroutines,\n    decorate your test methods with `tornado.testing.gen_test` instead\n    of `tornado.gen.coroutine`.\n\n    This class also provides the (deprecated) `stop()` and `wait()`\n    methods for a more manual style of testing. The test method itself\n    must call ``self.wait()``, and asynchronous callbacks should call\n    ``self.stop()`` to signal completion.\n\n    By default, a new `.IOLoop` is constructed for each test and is available\n    as ``self.io_loop``.  If the code being tested requires a\n    global `.IOLoop`, subclasses should override `get_new_ioloop` to return it.\n\n    The `.IOLoop`'s ``start`` and ``stop`` methods should not be\n    called directly.  Instead, use `self.stop <stop>` and `self.wait\n    <wait>`.  Arguments passed to ``self.stop`` are returned from\n    ``self.wait``.  It is possible to have multiple ``wait``/``stop``\n    cycles in the same test.\n\n    Example::\n\n        # This test uses coroutine style.\n        class MyTestCase(AsyncTestCase):\n            @tornado.testing.gen_test\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                response = yield client.fetch(\"http://www.tornadoweb.org\")\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n\n        # This test uses argument passing between self.stop and self.wait.\n        class MyTestCase2(AsyncTestCase):\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                client.fetch(\"http://www.tornadoweb.org/\", self.stop)\n                response = self.wait()\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n    \"\"\"\n\n    def __init__(self, methodName: str = \"runTest\") -> None:\n        super().__init__(methodName)\n        self.__stopped = False\n        self.__running = False\n        self.__failure = None  # type: Optional[_ExcInfoTuple]\n        self.__stop_args = None  # type: Any\n        self.__timeout = None  # type: Optional[object]\n\n        # It's easy to forget the @gen_test decorator, but if you do\n        # the test will silently be ignored because nothing will consume\n        # the generator.  Replace the test method with a wrapper that will\n        # make sure it's not an undecorated generator.\n        setattr(self, methodName, _TestMethodWrapper(getattr(self, methodName)))\n\n        # Not used in this class itself, but used by @gen_test\n        self._test_generator = None  # type: Optional[Union[Generator, Coroutine]]\n\n    def setUp(self) -> None:\n        setup_with_context_manager(self, warnings.catch_warnings())\n        warnings.filterwarnings(\n            \"ignore\",\n            message=\"There is no current event loop\",\n            category=DeprecationWarning,\n            module=r\"tornado\\..*\",\n        )\n        super().setUp()\n        # NOTE: this code attempts to navigate deprecation warnings introduced\n        # in Python 3.10. The idea of an implicit current event loop is\n        # deprecated in that version, with the intention that tests like this\n        # explicitly create a new event loop and run on it. However, other\n        # packages such as pytest-asyncio (as of version 0.16.0) still rely on\n        # the implicit current event loop and we want to be compatible with them\n        # (even when run on 3.10, but not, of course, on the future version of\n        # python that removes the get/set_event_loop methods completely).\n        #\n        # Deprecation warnings were introduced inconsistently:\n        # asyncio.get_event_loop warns, but\n        # asyncio.get_event_loop_policy().get_event_loop does not. Similarly,\n        # none of the set_event_loop methods warn, although comments on\n        # https://bugs.python.org/issue39529 indicate that they are also\n        # intended for future removal.\n        #\n        # Therefore, we first attempt to access the event loop with the\n        # (non-warning) policy method, and if it fails, fall back to creating a\n        # new event loop. We do not have effective test coverage of the\n        # new event loop case; this will have to be watched when/if\n        # get_event_loop is actually removed.\n        self.should_close_asyncio_loop = False\n        try:\n            self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()\n        except Exception:\n            self.asyncio_loop = asyncio.new_event_loop()\n            self.should_close_asyncio_loop = True\n\n        async def get_loop() -> IOLoop:\n            return self.get_new_ioloop()\n\n        self.io_loop = self.asyncio_loop.run_until_complete(get_loop())\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.make_current()\n\n    def tearDown(self) -> None:\n        # Native coroutines tend to produce warnings if they're not\n        # allowed to run to completion. It's difficult to ensure that\n        # this always happens in tests, so cancel any tasks that are\n        # still pending by the time we get here.\n        asyncio_loop = self.io_loop.asyncio_loop  # type: ignore\n        if hasattr(asyncio, \"all_tasks\"):  # py37\n            tasks = asyncio.all_tasks(asyncio_loop)  # type: ignore\n        else:\n            tasks = asyncio.Task.all_tasks(asyncio_loop)\n        # Tasks that are done may still appear here and may contain\n        # non-cancellation exceptions, so filter them out.\n        tasks = [t for t in tasks if not t.done()]  # type: ignore\n        for t in tasks:\n            t.cancel()\n        # Allow the tasks to run and finalize themselves (which means\n        # raising a CancelledError inside the coroutine). This may\n        # just transform the \"task was destroyed but it is pending\"\n        # warning into a \"uncaught CancelledError\" warning, but\n        # catching CancelledErrors in coroutines that may leak is\n        # simpler than ensuring that no coroutines leak.\n        if tasks:\n            done, pending = self.io_loop.run_sync(lambda: asyncio.wait(tasks))\n            assert not pending\n            # If any task failed with anything but a CancelledError, raise it.\n            for f in done:\n                try:\n                    f.result()\n                except asyncio.CancelledError:\n                    pass\n\n        # Clean up Subprocess, so it can be used again with a new ioloop.\n        Subprocess.uninitialize()\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.clear_current()\n        if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):\n            # Try to clean up any file descriptors left open in the ioloop.\n            # This avoids leaks, especially when tests are run repeatedly\n            # in the same process with autoreload (because curl does not\n            # set FD_CLOEXEC on its file descriptors)\n            self.io_loop.close(all_fds=True)\n        if self.should_close_asyncio_loop:\n            self.asyncio_loop.close()\n        super().tearDown()\n        # In case an exception escaped or the StackContext caught an exception\n        # when there wasn't a wait() to re-raise it, do so here.\n        # This is our last chance to raise an exception in a way that the\n        # unittest machinery understands.\n        self.__rethrow()\n\n    def get_new_ioloop(self) -> IOLoop:\n        \"\"\"Returns the `.IOLoop` to use for this test.\n\n        By default, a new `.IOLoop` is created for each test.\n        Subclasses may override this method to return\n        `.IOLoop.current()` if it is not appropriate to use a new\n        `.IOLoop` in each tests (for example, if there are global\n        singletons using the default `.IOLoop`) or if a per-test event\n        loop is being provided by another system (such as\n        ``pytest-asyncio``).\n        \"\"\"\n        return IOLoop(make_current=False)\n\n    def _handle_exception(\n        self, typ: Type[Exception], value: Exception, tb: TracebackType\n    ) -> bool:\n        if self.__failure is None:\n            self.__failure = (typ, value, tb)\n        else:\n            app_log.error(\n                \"multiple unhandled exceptions in test\", exc_info=(typ, value, tb)\n            )\n        self.stop()\n        return True\n\n    def __rethrow(self) -> None:\n        if self.__failure is not None:\n            failure = self.__failure\n            self.__failure = None\n            raise_exc_info(failure)\n\n    def run(\n        self, result: Optional[unittest.TestResult] = None\n    ) -> Optional[unittest.TestResult]:\n        ret = super().run(result)\n        # As a last resort, if an exception escaped super.run() and wasn't\n        # re-raised in tearDown, raise it here.  This will cause the\n        # unittest run to fail messily, but that's better than silently\n        # ignoring an error.\n        self.__rethrow()\n        return ret\n\n    def stop(self, _arg: Any = None, **kwargs: Any) -> None:\n        \"\"\"Stops the `.IOLoop`, causing one pending (or future) call to `wait()`\n        to return.\n\n        Keyword arguments or a single positional argument passed to `stop()` are\n        saved and will be returned by `wait()`.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        assert _arg is None or not kwargs\n        self.__stop_args = kwargs or _arg\n        if self.__running:\n            self.io_loop.stop()\n            self.__running = False\n        self.__stopped = True\n\n    def wait(\n        self,\n        condition: Optional[Callable[..., bool]] = None,\n        timeout: Optional[float] = None,\n    ) -> Any:\n        \"\"\"Runs the `.IOLoop` until stop is called or timeout has passed.\n\n        In the event of a timeout, an exception will be thrown. The\n        default timeout is 5 seconds; it may be overridden with a\n        ``timeout`` keyword argument or globally with the\n        ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        If ``condition`` is not ``None``, the `.IOLoop` will be restarted\n        after `stop()` until ``condition()`` returns ``True``.\n\n        .. versionchanged:: 3.1\n           Added the ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        if timeout is None:\n            timeout = get_async_test_timeout()\n\n        if not self.__stopped:\n            if timeout:\n\n                def timeout_func() -> None:\n                    try:\n                        raise self.failureException(\n                            \"Async operation timed out after %s seconds\" % timeout\n                        )\n                    except Exception:\n                        self.__failure = sys.exc_info()\n                    self.stop()\n\n                self.__timeout = self.io_loop.add_timeout(\n                    self.io_loop.time() + timeout, timeout_func\n                )\n            while True:\n                self.__running = True\n                self.io_loop.start()\n                if self.__failure is not None or condition is None or condition():\n                    break\n            if self.__timeout is not None:\n                self.io_loop.remove_timeout(self.__timeout)\n                self.__timeout = None\n        assert self.__stopped\n        self.__stopped = False\n        self.__rethrow()\n        result = self.__stop_args\n        self.__stop_args = None\n        return result",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 120,
                                "end_lineno": 393,
                                "parent": null
                            }
                        },
                        {
                            "file_name": "tornado/testing.py",
                            "type": "class",
                            "name": "AsyncTestCase",
                            "content": "class AsyncTestCase(unittest.TestCase):\n    \"\"\"`~unittest.TestCase` subclass for testing `.IOLoop`-based\n    asynchronous code.\n\n    The unittest framework is synchronous, so the test must be\n    complete by the time the test method returns. This means that\n    asynchronous code cannot be used in quite the same way as usual\n    and must be adapted to fit. To write your tests with coroutines,\n    decorate your test methods with `tornado.testing.gen_test` instead\n    of `tornado.gen.coroutine`.\n\n    This class also provides the (deprecated) `stop()` and `wait()`\n    methods for a more manual style of testing. The test method itself\n    must call ``self.wait()``, and asynchronous callbacks should call\n    ``self.stop()`` to signal completion.\n\n    By default, a new `.IOLoop` is constructed for each test and is available\n    as ``self.io_loop``.  If the code being tested requires a\n    global `.IOLoop`, subclasses should override `get_new_ioloop` to return it.\n\n    The `.IOLoop`'s ``start`` and ``stop`` methods should not be\n    called directly.  Instead, use `self.stop <stop>` and `self.wait\n    <wait>`.  Arguments passed to ``self.stop`` are returned from\n    ``self.wait``.  It is possible to have multiple ``wait``/``stop``\n    cycles in the same test.\n\n    Example::\n\n        # This test uses coroutine style.\n        class MyTestCase(AsyncTestCase):\n            @tornado.testing.gen_test\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                response = yield client.fetch(\"http://www.tornadoweb.org\")\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n\n        # This test uses argument passing between self.stop and self.wait.\n        class MyTestCase2(AsyncTestCase):\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                client.fetch(\"http://www.tornadoweb.org/\", self.stop)\n                response = self.wait()\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n    \"\"\"\n\n    def __init__(self, methodName: str = \"runTest\") -> None:\n        super().__init__(methodName)\n        self.__stopped = False\n        self.__running = False\n        self.__failure = None  # type: Optional[_ExcInfoTuple]\n        self.__stop_args = None  # type: Any\n        self.__timeout = None  # type: Optional[object]\n\n        # It's easy to forget the @gen_test decorator, but if you do\n        # the test will silently be ignored because nothing will consume\n        # the generator.  Replace the test method with a wrapper that will\n        # make sure it's not an undecorated generator.\n        setattr(self, methodName, _TestMethodWrapper(getattr(self, methodName)))\n\n        # Not used in this class itself, but used by @gen_test\n        self._test_generator = None  # type: Optional[Union[Generator, Coroutine]]\n\n    def setUp(self) -> None:\n        setup_with_context_manager(self, warnings.catch_warnings())\n        warnings.filterwarnings(\n            \"ignore\",\n            message=\"There is no current event loop\",\n            category=DeprecationWarning,\n            module=r\"tornado\\..*\",\n        )\n        super().setUp()\n        # NOTE: this code attempts to navigate deprecation warnings introduced\n        # in Python 3.10. The idea of an implicit current event loop is\n        # deprecated in that version, with the intention that tests like this\n        # explicitly create a new event loop and run on it. However, other\n        # packages such as pytest-asyncio (as of version 0.16.0) still rely on\n        # the implicit current event loop and we want to be compatible with them\n        # (even when run on 3.10, but not, of course, on the future version of\n        # python that removes the get/set_event_loop methods completely).\n        #\n        # Deprecation warnings were introduced inconsistently:\n        # asyncio.get_event_loop warns, but\n        # asyncio.get_event_loop_policy().get_event_loop does not. Similarly,\n        # none of the set_event_loop methods warn, although comments on\n        # https://bugs.python.org/issue39529 indicate that they are also\n        # intended for future removal.\n        #\n        # Therefore, we first attempt to access the event loop with the\n        # (non-warning) policy method, and if it fails, fall back to creating a\n        # new event loop. We do not have effective test coverage of the\n        # new event loop case; this will have to be watched when/if\n        # get_event_loop is actually removed.\n        self.should_close_asyncio_loop = False\n        try:\n            self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()\n        except Exception:\n            self.asyncio_loop = asyncio.new_event_loop()\n            self.should_close_asyncio_loop = True\n\n        async def get_loop() -> IOLoop:\n            return self.get_new_ioloop()\n\n        self.io_loop = self.asyncio_loop.run_until_complete(get_loop())\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.make_current()\n\n    def tearDown(self) -> None:\n        # Native coroutines tend to produce warnings if they're not\n        # allowed to run to completion. It's difficult to ensure that\n        # this always happens in tests, so cancel any tasks that are\n        # still pending by the time we get here.\n        asyncio_loop = self.io_loop.asyncio_loop  # type: ignore\n        if hasattr(asyncio, \"all_tasks\"):  # py37\n            tasks = asyncio.all_tasks(asyncio_loop)  # type: ignore\n        else:\n            tasks = asyncio.Task.all_tasks(asyncio_loop)\n        # Tasks that are done may still appear here and may contain\n        # non-cancellation exceptions, so filter them out.\n        tasks = [t for t in tasks if not t.done()]  # type: ignore\n        for t in tasks:\n            t.cancel()\n        # Allow the tasks to run and finalize themselves (which means\n        # raising a CancelledError inside the coroutine). This may\n        # just transform the \"task was destroyed but it is pending\"\n        # warning into a \"uncaught CancelledError\" warning, but\n        # catching CancelledErrors in coroutines that may leak is\n        # simpler than ensuring that no coroutines leak.\n        if tasks:\n            done, pending = self.io_loop.run_sync(lambda: asyncio.wait(tasks))\n            assert not pending\n            # If any task failed with anything but a CancelledError, raise it.\n            for f in done:\n                try:\n                    f.result()\n                except asyncio.CancelledError:\n                    pass\n\n        # Clean up Subprocess, so it can be used again with a new ioloop.\n        Subprocess.uninitialize()\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.clear_current()\n        if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):\n            # Try to clean up any file descriptors left open in the ioloop.\n            # This avoids leaks, especially when tests are run repeatedly\n            # in the same process with autoreload (because curl does not\n            # set FD_CLOEXEC on its file descriptors)\n            self.io_loop.close(all_fds=True)\n        if self.should_close_asyncio_loop:\n            self.asyncio_loop.close()\n        super().tearDown()\n        # In case an exception escaped or the StackContext caught an exception\n        # when there wasn't a wait() to re-raise it, do so here.\n        # This is our last chance to raise an exception in a way that the\n        # unittest machinery understands.\n        self.__rethrow()\n\n    def get_new_ioloop(self) -> IOLoop:\n        \"\"\"Returns the `.IOLoop` to use for this test.\n\n        By default, a new `.IOLoop` is created for each test.\n        Subclasses may override this method to return\n        `.IOLoop.current()` if it is not appropriate to use a new\n        `.IOLoop` in each tests (for example, if there are global\n        singletons using the default `.IOLoop`) or if a per-test event\n        loop is being provided by another system (such as\n        ``pytest-asyncio``).\n        \"\"\"\n        return IOLoop(make_current=False)\n\n    def _handle_exception(\n        self, typ: Type[Exception], value: Exception, tb: TracebackType\n    ) -> bool:\n        if self.__failure is None:\n            self.__failure = (typ, value, tb)\n        else:\n            app_log.error(\n                \"multiple unhandled exceptions in test\", exc_info=(typ, value, tb)\n            )\n        self.stop()\n        return True\n\n    def __rethrow(self) -> None:\n        if self.__failure is not None:\n            failure = self.__failure\n            self.__failure = None\n            raise_exc_info(failure)\n\n    def run(\n        self, result: Optional[unittest.TestResult] = None\n    ) -> Optional[unittest.TestResult]:\n        ret = super().run(result)\n        # As a last resort, if an exception escaped super.run() and wasn't\n        # re-raised in tearDown, raise it here.  This will cause the\n        # unittest run to fail messily, but that's better than silently\n        # ignoring an error.\n        self.__rethrow()\n        return ret\n\n    def stop(self, _arg: Any = None, **kwargs: Any) -> None:\n        \"\"\"Stops the `.IOLoop`, causing one pending (or future) call to `wait()`\n        to return.\n\n        Keyword arguments or a single positional argument passed to `stop()` are\n        saved and will be returned by `wait()`.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        assert _arg is None or not kwargs\n        self.__stop_args = kwargs or _arg\n        if self.__running:\n            self.io_loop.stop()\n            self.__running = False\n        self.__stopped = True\n\n    def wait(\n        self,\n        condition: Optional[Callable[..., bool]] = None,\n        timeout: Optional[float] = None,\n    ) -> Any:\n        \"\"\"Runs the `.IOLoop` until stop is called or timeout has passed.\n\n        In the event of a timeout, an exception will be thrown. The\n        default timeout is 5 seconds; it may be overridden with a\n        ``timeout`` keyword argument or globally with the\n        ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        If ``condition`` is not ``None``, the `.IOLoop` will be restarted\n        after `stop()` until ``condition()`` returns ``True``.\n\n        .. versionchanged:: 3.1\n           Added the ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        if timeout is None:\n            timeout = get_async_test_timeout()\n\n        if not self.__stopped:\n            if timeout:\n\n                def timeout_func() -> None:\n                    try:\n                        raise self.failureException(\n                            \"Async operation timed out after %s seconds\" % timeout\n                        )\n                    except Exception:\n                        self.__failure = sys.exc_info()\n                    self.stop()\n\n                self.__timeout = self.io_loop.add_timeout(\n                    self.io_loop.time() + timeout, timeout_func\n                )\n            while True:\n                self.__running = True\n                self.io_loop.start()\n                if self.__failure is not None or condition is None or condition():\n                    break\n            if self.__timeout is not None:\n                self.io_loop.remove_timeout(self.__timeout)\n                self.__timeout = None\n        assert self.__stopped\n        self.__stopped = False\n        self.__rethrow()\n        result = self.__stop_args\n        self.__stop_args = None\n        return result",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 120,
                            "end_lineno": 393,
                            "parent": null
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 259,
                            "length": 17
                        },
                        "new_range": {
                            "start": 219,
                            "length": 13
                        },
                        "section": "class AsyncTestCase(unittest.TestCase):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "        # Clean up Subprocess, so it can be used again with a new ioloop.",
                                "type": "context"
                            },
                            {
                                "content": "        Subprocess.uninitialize()",
                                "type": "context"
                            },
                            {
                                "content": "        with warnings.catch_warnings():",
                                "type": "deleted"
                            },
                            {
                                "content": "            warnings.simplefilter(\"ignore\", DeprecationWarning)",
                                "type": "deleted"
                            },
                            {
                                "content": "            self.io_loop.clear_current()",
                                "type": "deleted"
                            },
                            {
                                "content": "        self.io_loop.clear_current()",
                                "type": "added"
                            },
                            {
                                "content": "        if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):",
                                "type": "context"
                            },
                            {
                                "content": "            # Try to clean up any file descriptors left open in the ioloop.",
                                "type": "context"
                            },
                            {
                                "content": "            # This avoids leaks, especially when tests are run repeatedly",
                                "type": "context"
                            },
                            {
                                "content": "            # in the same process with autoreload (because curl does not",
                                "type": "context"
                            },
                            {
                                "content": "            # set FD_CLOEXEC on its file descriptors)",
                                "type": "context"
                            },
                            {
                                "content": "            self.io_loop.close(all_fds=True)",
                                "type": "context"
                            },
                            {
                                "content": "        if self.should_close_asyncio_loop:",
                                "type": "deleted"
                            },
                            {
                                "content": "            self.asyncio_loop.close()",
                                "type": "deleted"
                            },
                            {
                                "content": "        super().tearDown()",
                                "type": "context"
                            },
                            {
                                "content": "        # In case an exception escaped or the StackContext caught an exception",
                                "type": "context"
                            },
                            {
                                "content": "        # when there wasn't a wait() to re-raise it, do so here.",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/testing.py",
                            "type": "method",
                            "name": "AsyncTestCase.tearDown",
                            "content": "class AsyncTestCase(unittest.TestCase):\n    \"\"\"`~unittest.TestCase` subclass for testing `.IOLoop`-based\n    asynchronous code.\n\n    The unittest framework is synchronous, so the test must be\n    complete by the time the test method returns. This means that\n    asynchronous code cannot be used in quite the same way as usual\n    and must be adapted to fit. To write your tests with coroutines,\n    decorate your test methods with `tornado.testing.gen_test` instead\n    of `tornado.gen.coroutine`.\n\n    This class also provides the (deprecated) `stop()` and `wait()`\n    methods for a more manual style of testing. The test method itself\n    must call ``self.wait()``, and asynchronous callbacks should call\n    ``self.stop()`` to signal completion.\n\n    By default, a new `.IOLoop` is constructed for each test and is available\n    as ``self.io_loop``.  If the code being tested requires a\n    global `.IOLoop`, subclasses should override `get_new_ioloop` to return it.\n\n    The `.IOLoop`'s ``start`` and ``stop`` methods should not be\n    called directly.  Instead, use `self.stop <stop>` and `self.wait\n    <wait>`.  Arguments passed to ``self.stop`` are returned from\n    ``self.wait``.  It is possible to have multiple ``wait``/``stop``\n    cycles in the same test.\n\n    Example::\n\n        # This test uses coroutine style.\n        class MyTestCase(AsyncTestCase):\n            @tornado.testing.gen_test\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                response = yield client.fetch(\"http://www.tornadoweb.org\")\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n\n        # This test uses argument passing between self.stop and self.wait.\n        class MyTestCase2(AsyncTestCase):\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                client.fetch(\"http://www.tornadoweb.org/\", self.stop)\n                response = self.wait()\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n    \"\"\"\n\n    def __init__(self, methodName: str = \"runTest\") -> None:\n        super().__init__(methodName)\n        self.__stopped = False\n        self.__running = False\n        self.__failure = None  # type: Optional[_ExcInfoTuple]\n        self.__stop_args = None  # type: Any\n        self.__timeout = None  # type: Optional[object]\n\n        # It's easy to forget the @gen_test decorator, but if you do\n        # the test will silently be ignored because nothing will consume\n        # the generator.  Replace the test method with a wrapper that will\n        # make sure it's not an undecorated generator.\n        setattr(self, methodName, _TestMethodWrapper(getattr(self, methodName)))\n\n        # Not used in this class itself, but used by @gen_test\n        self._test_generator = None  # type: Optional[Union[Generator, Coroutine]]\n\n    def setUp(self) -> None:\n        setup_with_context_manager(self, warnings.catch_warnings())\n        warnings.filterwarnings(\n            \"ignore\",\n            message=\"There is no current event loop\",\n            category=DeprecationWarning,\n            module=r\"tornado\\..*\",\n        )\n        super().setUp()\n        # NOTE: this code attempts to navigate deprecation warnings introduced\n        # in Python 3.10. The idea of an implicit current event loop is\n        # deprecated in that version, with the intention that tests like this\n        # explicitly create a new event loop and run on it. However, other\n        # packages such as pytest-asyncio (as of version 0.16.0) still rely on\n        # the implicit current event loop and we want to be compatible with them\n        # (even when run on 3.10, but not, of course, on the future version of\n        # python that removes the get/set_event_loop methods completely).\n        #\n        # Deprecation warnings were introduced inconsistently:\n        # asyncio.get_event_loop warns, but\n        # asyncio.get_event_loop_policy().get_event_loop does not. Similarly,\n        # none of the set_event_loop methods warn, although comments on\n        # https://bugs.python.org/issue39529 indicate that they are also\n        # intended for future removal.\n        #\n        # Therefore, we first attempt to access the event loop with the\n        # (non-warning) policy method, and if it fails, fall back to creating a\n        # new event loop. We do not have effective test coverage of the\n        # new event loop case; this will have to be watched when/if\n        # get_event_loop is actually removed.\n        self.should_close_asyncio_loop = False\n        try:\n            self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()\n        except Exception:\n            self.asyncio_loop = asyncio.new_event_loop()\n            self.should_close_asyncio_loop = True\n\n        async def get_loop() -> IOLoop:\n            return self.get_new_ioloop()\n\n        self.io_loop = self.asyncio_loop.run_until_complete(get_loop())\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.make_current()\n\n    def tearDown(self) -> None:\n        # Native coroutines tend to produce warnings if they're not\n        # allowed to run to completion. It's difficult to ensure that\n        # this always happens in tests, so cancel any tasks that are\n        # still pending by the time we get here.\n        asyncio_loop = self.io_loop.asyncio_loop  # type: ignore\n        if hasattr(asyncio, \"all_tasks\"):  # py37\n            tasks = asyncio.all_tasks(asyncio_loop)  # type: ignore\n        else:\n            tasks = asyncio.Task.all_tasks(asyncio_loop)\n        # Tasks that are done may still appear here and may contain\n        # non-cancellation exceptions, so filter them out.\n        tasks = [t for t in tasks if not t.done()]  # type: ignore\n        for t in tasks:\n            t.cancel()\n        # Allow the tasks to run and finalize themselves (which means\n        # raising a CancelledError inside the coroutine). This may\n        # just transform the \"task was destroyed but it is pending\"\n        # warning into a \"uncaught CancelledError\" warning, but\n        # catching CancelledErrors in coroutines that may leak is\n        # simpler than ensuring that no coroutines leak.\n        if tasks:\n            done, pending = self.io_loop.run_sync(lambda: asyncio.wait(tasks))\n            assert not pending\n            # If any task failed with anything but a CancelledError, raise it.\n            for f in done:\n                try:\n                    f.result()\n                except asyncio.CancelledError:\n                    pass\n\n        # Clean up Subprocess, so it can be used again with a new ioloop.\n        Subprocess.uninitialize()\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.clear_current()\n        if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):\n            # Try to clean up any file descriptors left open in the ioloop.\n            # This avoids leaks, especially when tests are run repeatedly\n            # in the same process with autoreload (because curl does not\n            # set FD_CLOEXEC on its file descriptors)\n            self.io_loop.close(all_fds=True)\n        if self.should_close_asyncio_loop:\n            self.asyncio_loop.close()\n        super().tearDown()\n        # In case an exception escaped or the StackContext caught an exception\n        # when there wasn't a wait() to re-raise it, do so here.\n        # This is our last chance to raise an exception in a way that the\n        # unittest machinery understands.\n        self.__rethrow()\n\n    def get_new_ioloop(self) -> IOLoop:\n        \"\"\"Returns the `.IOLoop` to use for this test.\n\n        By default, a new `.IOLoop` is created for each test.\n        Subclasses may override this method to return\n        `.IOLoop.current()` if it is not appropriate to use a new\n        `.IOLoop` in each tests (for example, if there are global\n        singletons using the default `.IOLoop`) or if a per-test event\n        loop is being provided by another system (such as\n        ``pytest-asyncio``).\n        \"\"\"\n        return IOLoop(make_current=False)\n\n    def _handle_exception(\n        self, typ: Type[Exception], value: Exception, tb: TracebackType\n    ) -> bool:\n        if self.__failure is None:\n            self.__failure = (typ, value, tb)\n        else:\n            app_log.error(\n                \"multiple unhandled exceptions in test\", exc_info=(typ, value, tb)\n            )\n        self.stop()\n        return True\n\n    def __rethrow(self) -> None:\n        if self.__failure is not None:\n            failure = self.__failure\n            self.__failure = None\n            raise_exc_info(failure)\n\n    def run(\n        self, result: Optional[unittest.TestResult] = None\n    ) -> Optional[unittest.TestResult]:\n        ret = super().run(result)\n        # As a last resort, if an exception escaped super.run() and wasn't\n        # re-raised in tearDown, raise it here.  This will cause the\n        # unittest run to fail messily, but that's better than silently\n        # ignoring an error.\n        self.__rethrow()\n        return ret\n\n    def stop(self, _arg: Any = None, **kwargs: Any) -> None:\n        \"\"\"Stops the `.IOLoop`, causing one pending (or future) call to `wait()`\n        to return.\n\n        Keyword arguments or a single positional argument passed to `stop()` are\n        saved and will be returned by `wait()`.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        assert _arg is None or not kwargs\n        self.__stop_args = kwargs or _arg\n        if self.__running:\n            self.io_loop.stop()\n            self.__running = False\n        self.__stopped = True\n\n    def wait(\n        self,\n        condition: Optional[Callable[..., bool]] = None,\n        timeout: Optional[float] = None,\n    ) -> Any:\n        \"\"\"Runs the `.IOLoop` until stop is called or timeout has passed.\n\n        In the event of a timeout, an exception will be thrown. The\n        default timeout is 5 seconds; it may be overridden with a\n        ``timeout`` keyword argument or globally with the\n        ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        If ``condition`` is not ``None``, the `.IOLoop` will be restarted\n        after `stop()` until ``condition()`` returns ``True``.\n\n        .. versionchanged:: 3.1\n           Added the ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        if timeout is None:\n            timeout = get_async_test_timeout()\n\n        if not self.__stopped:\n            if timeout:\n\n                def timeout_func() -> None:\n                    try:\n                        raise self.failureException(\n                            \"Async operation timed out after %s seconds\" % timeout\n                        )\n                    except Exception:\n                        self.__failure = sys.exc_info()\n                    self.stop()\n\n                self.__timeout = self.io_loop.add_timeout(\n                    self.io_loop.time() + timeout, timeout_func\n                )\n            while True:\n                self.__running = True\n                self.io_loop.start()\n                if self.__failure is not None or condition is None or condition():\n                    break\n            if self.__timeout is not None:\n                self.io_loop.remove_timeout(self.__timeout)\n                self.__timeout = None\n        assert self.__stopped\n        self.__stopped = False\n        self.__rethrow()\n        result = self.__stop_args\n        self.__stop_args = None\n        return result",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 229,
                            "end_lineno": 278,
                            "parent": {
                                "file_name": "tornado/testing.py",
                                "type": "class",
                                "name": "AsyncTestCase",
                                "content": "class AsyncTestCase(unittest.TestCase):\n    \"\"\"`~unittest.TestCase` subclass for testing `.IOLoop`-based\n    asynchronous code.\n\n    The unittest framework is synchronous, so the test must be\n    complete by the time the test method returns. This means that\n    asynchronous code cannot be used in quite the same way as usual\n    and must be adapted to fit. To write your tests with coroutines,\n    decorate your test methods with `tornado.testing.gen_test` instead\n    of `tornado.gen.coroutine`.\n\n    This class also provides the (deprecated) `stop()` and `wait()`\n    methods for a more manual style of testing. The test method itself\n    must call ``self.wait()``, and asynchronous callbacks should call\n    ``self.stop()`` to signal completion.\n\n    By default, a new `.IOLoop` is constructed for each test and is available\n    as ``self.io_loop``.  If the code being tested requires a\n    global `.IOLoop`, subclasses should override `get_new_ioloop` to return it.\n\n    The `.IOLoop`'s ``start`` and ``stop`` methods should not be\n    called directly.  Instead, use `self.stop <stop>` and `self.wait\n    <wait>`.  Arguments passed to ``self.stop`` are returned from\n    ``self.wait``.  It is possible to have multiple ``wait``/``stop``\n    cycles in the same test.\n\n    Example::\n\n        # This test uses coroutine style.\n        class MyTestCase(AsyncTestCase):\n            @tornado.testing.gen_test\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                response = yield client.fetch(\"http://www.tornadoweb.org\")\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n\n        # This test uses argument passing between self.stop and self.wait.\n        class MyTestCase2(AsyncTestCase):\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                client.fetch(\"http://www.tornadoweb.org/\", self.stop)\n                response = self.wait()\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n    \"\"\"\n\n    def __init__(self, methodName: str = \"runTest\") -> None:\n        super().__init__(methodName)\n        self.__stopped = False\n        self.__running = False\n        self.__failure = None  # type: Optional[_ExcInfoTuple]\n        self.__stop_args = None  # type: Any\n        self.__timeout = None  # type: Optional[object]\n\n        # It's easy to forget the @gen_test decorator, but if you do\n        # the test will silently be ignored because nothing will consume\n        # the generator.  Replace the test method with a wrapper that will\n        # make sure it's not an undecorated generator.\n        setattr(self, methodName, _TestMethodWrapper(getattr(self, methodName)))\n\n        # Not used in this class itself, but used by @gen_test\n        self._test_generator = None  # type: Optional[Union[Generator, Coroutine]]\n\n    def setUp(self) -> None:\n        setup_with_context_manager(self, warnings.catch_warnings())\n        warnings.filterwarnings(\n            \"ignore\",\n            message=\"There is no current event loop\",\n            category=DeprecationWarning,\n            module=r\"tornado\\..*\",\n        )\n        super().setUp()\n        # NOTE: this code attempts to navigate deprecation warnings introduced\n        # in Python 3.10. The idea of an implicit current event loop is\n        # deprecated in that version, with the intention that tests like this\n        # explicitly create a new event loop and run on it. However, other\n        # packages such as pytest-asyncio (as of version 0.16.0) still rely on\n        # the implicit current event loop and we want to be compatible with them\n        # (even when run on 3.10, but not, of course, on the future version of\n        # python that removes the get/set_event_loop methods completely).\n        #\n        # Deprecation warnings were introduced inconsistently:\n        # asyncio.get_event_loop warns, but\n        # asyncio.get_event_loop_policy().get_event_loop does not. Similarly,\n        # none of the set_event_loop methods warn, although comments on\n        # https://bugs.python.org/issue39529 indicate that they are also\n        # intended for future removal.\n        #\n        # Therefore, we first attempt to access the event loop with the\n        # (non-warning) policy method, and if it fails, fall back to creating a\n        # new event loop. We do not have effective test coverage of the\n        # new event loop case; this will have to be watched when/if\n        # get_event_loop is actually removed.\n        self.should_close_asyncio_loop = False\n        try:\n            self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()\n        except Exception:\n            self.asyncio_loop = asyncio.new_event_loop()\n            self.should_close_asyncio_loop = True\n\n        async def get_loop() -> IOLoop:\n            return self.get_new_ioloop()\n\n        self.io_loop = self.asyncio_loop.run_until_complete(get_loop())\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.make_current()\n\n    def tearDown(self) -> None:\n        # Native coroutines tend to produce warnings if they're not\n        # allowed to run to completion. It's difficult to ensure that\n        # this always happens in tests, so cancel any tasks that are\n        # still pending by the time we get here.\n        asyncio_loop = self.io_loop.asyncio_loop  # type: ignore\n        if hasattr(asyncio, \"all_tasks\"):  # py37\n            tasks = asyncio.all_tasks(asyncio_loop)  # type: ignore\n        else:\n            tasks = asyncio.Task.all_tasks(asyncio_loop)\n        # Tasks that are done may still appear here and may contain\n        # non-cancellation exceptions, so filter them out.\n        tasks = [t for t in tasks if not t.done()]  # type: ignore\n        for t in tasks:\n            t.cancel()\n        # Allow the tasks to run and finalize themselves (which means\n        # raising a CancelledError inside the coroutine). This may\n        # just transform the \"task was destroyed but it is pending\"\n        # warning into a \"uncaught CancelledError\" warning, but\n        # catching CancelledErrors in coroutines that may leak is\n        # simpler than ensuring that no coroutines leak.\n        if tasks:\n            done, pending = self.io_loop.run_sync(lambda: asyncio.wait(tasks))\n            assert not pending\n            # If any task failed with anything but a CancelledError, raise it.\n            for f in done:\n                try:\n                    f.result()\n                except asyncio.CancelledError:\n                    pass\n\n        # Clean up Subprocess, so it can be used again with a new ioloop.\n        Subprocess.uninitialize()\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.clear_current()\n        if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):\n            # Try to clean up any file descriptors left open in the ioloop.\n            # This avoids leaks, especially when tests are run repeatedly\n            # in the same process with autoreload (because curl does not\n            # set FD_CLOEXEC on its file descriptors)\n            self.io_loop.close(all_fds=True)\n        if self.should_close_asyncio_loop:\n            self.asyncio_loop.close()\n        super().tearDown()\n        # In case an exception escaped or the StackContext caught an exception\n        # when there wasn't a wait() to re-raise it, do so here.\n        # This is our last chance to raise an exception in a way that the\n        # unittest machinery understands.\n        self.__rethrow()\n\n    def get_new_ioloop(self) -> IOLoop:\n        \"\"\"Returns the `.IOLoop` to use for this test.\n\n        By default, a new `.IOLoop` is created for each test.\n        Subclasses may override this method to return\n        `.IOLoop.current()` if it is not appropriate to use a new\n        `.IOLoop` in each tests (for example, if there are global\n        singletons using the default `.IOLoop`) or if a per-test event\n        loop is being provided by another system (such as\n        ``pytest-asyncio``).\n        \"\"\"\n        return IOLoop(make_current=False)\n\n    def _handle_exception(\n        self, typ: Type[Exception], value: Exception, tb: TracebackType\n    ) -> bool:\n        if self.__failure is None:\n            self.__failure = (typ, value, tb)\n        else:\n            app_log.error(\n                \"multiple unhandled exceptions in test\", exc_info=(typ, value, tb)\n            )\n        self.stop()\n        return True\n\n    def __rethrow(self) -> None:\n        if self.__failure is not None:\n            failure = self.__failure\n            self.__failure = None\n            raise_exc_info(failure)\n\n    def run(\n        self, result: Optional[unittest.TestResult] = None\n    ) -> Optional[unittest.TestResult]:\n        ret = super().run(result)\n        # As a last resort, if an exception escaped super.run() and wasn't\n        # re-raised in tearDown, raise it here.  This will cause the\n        # unittest run to fail messily, but that's better than silently\n        # ignoring an error.\n        self.__rethrow()\n        return ret\n\n    def stop(self, _arg: Any = None, **kwargs: Any) -> None:\n        \"\"\"Stops the `.IOLoop`, causing one pending (or future) call to `wait()`\n        to return.\n\n        Keyword arguments or a single positional argument passed to `stop()` are\n        saved and will be returned by `wait()`.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        assert _arg is None or not kwargs\n        self.__stop_args = kwargs or _arg\n        if self.__running:\n            self.io_loop.stop()\n            self.__running = False\n        self.__stopped = True\n\n    def wait(\n        self,\n        condition: Optional[Callable[..., bool]] = None,\n        timeout: Optional[float] = None,\n    ) -> Any:\n        \"\"\"Runs the `.IOLoop` until stop is called or timeout has passed.\n\n        In the event of a timeout, an exception will be thrown. The\n        default timeout is 5 seconds; it may be overridden with a\n        ``timeout`` keyword argument or globally with the\n        ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        If ``condition`` is not ``None``, the `.IOLoop` will be restarted\n        after `stop()` until ``condition()`` returns ``True``.\n\n        .. versionchanged:: 3.1\n           Added the ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        if timeout is None:\n            timeout = get_async_test_timeout()\n\n        if not self.__stopped:\n            if timeout:\n\n                def timeout_func() -> None:\n                    try:\n                        raise self.failureException(\n                            \"Async operation timed out after %s seconds\" % timeout\n                        )\n                    except Exception:\n                        self.__failure = sys.exc_info()\n                    self.stop()\n\n                self.__timeout = self.io_loop.add_timeout(\n                    self.io_loop.time() + timeout, timeout_func\n                )\n            while True:\n                self.__running = True\n                self.io_loop.start()\n                if self.__failure is not None or condition is None or condition():\n                    break\n            if self.__timeout is not None:\n                self.io_loop.remove_timeout(self.__timeout)\n                self.__timeout = None\n        assert self.__stopped\n        self.__stopped = False\n        self.__rethrow()\n        result = self.__stop_args\n        self.__stop_args = None\n        return result",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 120,
                                "end_lineno": 393,
                                "parent": null
                            }
                        },
                        {
                            "file_name": "tornado/testing.py",
                            "type": "class",
                            "name": "AsyncTestCase",
                            "content": "class AsyncTestCase(unittest.TestCase):\n    \"\"\"`~unittest.TestCase` subclass for testing `.IOLoop`-based\n    asynchronous code.\n\n    The unittest framework is synchronous, so the test must be\n    complete by the time the test method returns. This means that\n    asynchronous code cannot be used in quite the same way as usual\n    and must be adapted to fit. To write your tests with coroutines,\n    decorate your test methods with `tornado.testing.gen_test` instead\n    of `tornado.gen.coroutine`.\n\n    This class also provides the (deprecated) `stop()` and `wait()`\n    methods for a more manual style of testing. The test method itself\n    must call ``self.wait()``, and asynchronous callbacks should call\n    ``self.stop()`` to signal completion.\n\n    By default, a new `.IOLoop` is constructed for each test and is available\n    as ``self.io_loop``.  If the code being tested requires a\n    global `.IOLoop`, subclasses should override `get_new_ioloop` to return it.\n\n    The `.IOLoop`'s ``start`` and ``stop`` methods should not be\n    called directly.  Instead, use `self.stop <stop>` and `self.wait\n    <wait>`.  Arguments passed to ``self.stop`` are returned from\n    ``self.wait``.  It is possible to have multiple ``wait``/``stop``\n    cycles in the same test.\n\n    Example::\n\n        # This test uses coroutine style.\n        class MyTestCase(AsyncTestCase):\n            @tornado.testing.gen_test\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                response = yield client.fetch(\"http://www.tornadoweb.org\")\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n\n        # This test uses argument passing between self.stop and self.wait.\n        class MyTestCase2(AsyncTestCase):\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                client.fetch(\"http://www.tornadoweb.org/\", self.stop)\n                response = self.wait()\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n    \"\"\"\n\n    def __init__(self, methodName: str = \"runTest\") -> None:\n        super().__init__(methodName)\n        self.__stopped = False\n        self.__running = False\n        self.__failure = None  # type: Optional[_ExcInfoTuple]\n        self.__stop_args = None  # type: Any\n        self.__timeout = None  # type: Optional[object]\n\n        # It's easy to forget the @gen_test decorator, but if you do\n        # the test will silently be ignored because nothing will consume\n        # the generator.  Replace the test method with a wrapper that will\n        # make sure it's not an undecorated generator.\n        setattr(self, methodName, _TestMethodWrapper(getattr(self, methodName)))\n\n        # Not used in this class itself, but used by @gen_test\n        self._test_generator = None  # type: Optional[Union[Generator, Coroutine]]\n\n    def setUp(self) -> None:\n        setup_with_context_manager(self, warnings.catch_warnings())\n        warnings.filterwarnings(\n            \"ignore\",\n            message=\"There is no current event loop\",\n            category=DeprecationWarning,\n            module=r\"tornado\\..*\",\n        )\n        super().setUp()\n        # NOTE: this code attempts to navigate deprecation warnings introduced\n        # in Python 3.10. The idea of an implicit current event loop is\n        # deprecated in that version, with the intention that tests like this\n        # explicitly create a new event loop and run on it. However, other\n        # packages such as pytest-asyncio (as of version 0.16.0) still rely on\n        # the implicit current event loop and we want to be compatible with them\n        # (even when run on 3.10, but not, of course, on the future version of\n        # python that removes the get/set_event_loop methods completely).\n        #\n        # Deprecation warnings were introduced inconsistently:\n        # asyncio.get_event_loop warns, but\n        # asyncio.get_event_loop_policy().get_event_loop does not. Similarly,\n        # none of the set_event_loop methods warn, although comments on\n        # https://bugs.python.org/issue39529 indicate that they are also\n        # intended for future removal.\n        #\n        # Therefore, we first attempt to access the event loop with the\n        # (non-warning) policy method, and if it fails, fall back to creating a\n        # new event loop. We do not have effective test coverage of the\n        # new event loop case; this will have to be watched when/if\n        # get_event_loop is actually removed.\n        self.should_close_asyncio_loop = False\n        try:\n            self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()\n        except Exception:\n            self.asyncio_loop = asyncio.new_event_loop()\n            self.should_close_asyncio_loop = True\n\n        async def get_loop() -> IOLoop:\n            return self.get_new_ioloop()\n\n        self.io_loop = self.asyncio_loop.run_until_complete(get_loop())\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.make_current()\n\n    def tearDown(self) -> None:\n        # Native coroutines tend to produce warnings if they're not\n        # allowed to run to completion. It's difficult to ensure that\n        # this always happens in tests, so cancel any tasks that are\n        # still pending by the time we get here.\n        asyncio_loop = self.io_loop.asyncio_loop  # type: ignore\n        if hasattr(asyncio, \"all_tasks\"):  # py37\n            tasks = asyncio.all_tasks(asyncio_loop)  # type: ignore\n        else:\n            tasks = asyncio.Task.all_tasks(asyncio_loop)\n        # Tasks that are done may still appear here and may contain\n        # non-cancellation exceptions, so filter them out.\n        tasks = [t for t in tasks if not t.done()]  # type: ignore\n        for t in tasks:\n            t.cancel()\n        # Allow the tasks to run and finalize themselves (which means\n        # raising a CancelledError inside the coroutine). This may\n        # just transform the \"task was destroyed but it is pending\"\n        # warning into a \"uncaught CancelledError\" warning, but\n        # catching CancelledErrors in coroutines that may leak is\n        # simpler than ensuring that no coroutines leak.\n        if tasks:\n            done, pending = self.io_loop.run_sync(lambda: asyncio.wait(tasks))\n            assert not pending\n            # If any task failed with anything but a CancelledError, raise it.\n            for f in done:\n                try:\n                    f.result()\n                except asyncio.CancelledError:\n                    pass\n\n        # Clean up Subprocess, so it can be used again with a new ioloop.\n        Subprocess.uninitialize()\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.clear_current()\n        if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):\n            # Try to clean up any file descriptors left open in the ioloop.\n            # This avoids leaks, especially when tests are run repeatedly\n            # in the same process with autoreload (because curl does not\n            # set FD_CLOEXEC on its file descriptors)\n            self.io_loop.close(all_fds=True)\n        if self.should_close_asyncio_loop:\n            self.asyncio_loop.close()\n        super().tearDown()\n        # In case an exception escaped or the StackContext caught an exception\n        # when there wasn't a wait() to re-raise it, do so here.\n        # This is our last chance to raise an exception in a way that the\n        # unittest machinery understands.\n        self.__rethrow()\n\n    def get_new_ioloop(self) -> IOLoop:\n        \"\"\"Returns the `.IOLoop` to use for this test.\n\n        By default, a new `.IOLoop` is created for each test.\n        Subclasses may override this method to return\n        `.IOLoop.current()` if it is not appropriate to use a new\n        `.IOLoop` in each tests (for example, if there are global\n        singletons using the default `.IOLoop`) or if a per-test event\n        loop is being provided by another system (such as\n        ``pytest-asyncio``).\n        \"\"\"\n        return IOLoop(make_current=False)\n\n    def _handle_exception(\n        self, typ: Type[Exception], value: Exception, tb: TracebackType\n    ) -> bool:\n        if self.__failure is None:\n            self.__failure = (typ, value, tb)\n        else:\n            app_log.error(\n                \"multiple unhandled exceptions in test\", exc_info=(typ, value, tb)\n            )\n        self.stop()\n        return True\n\n    def __rethrow(self) -> None:\n        if self.__failure is not None:\n            failure = self.__failure\n            self.__failure = None\n            raise_exc_info(failure)\n\n    def run(\n        self, result: Optional[unittest.TestResult] = None\n    ) -> Optional[unittest.TestResult]:\n        ret = super().run(result)\n        # As a last resort, if an exception escaped super.run() and wasn't\n        # re-raised in tearDown, raise it here.  This will cause the\n        # unittest run to fail messily, but that's better than silently\n        # ignoring an error.\n        self.__rethrow()\n        return ret\n\n    def stop(self, _arg: Any = None, **kwargs: Any) -> None:\n        \"\"\"Stops the `.IOLoop`, causing one pending (or future) call to `wait()`\n        to return.\n\n        Keyword arguments or a single positional argument passed to `stop()` are\n        saved and will be returned by `wait()`.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        assert _arg is None or not kwargs\n        self.__stop_args = kwargs or _arg\n        if self.__running:\n            self.io_loop.stop()\n            self.__running = False\n        self.__stopped = True\n\n    def wait(\n        self,\n        condition: Optional[Callable[..., bool]] = None,\n        timeout: Optional[float] = None,\n    ) -> Any:\n        \"\"\"Runs the `.IOLoop` until stop is called or timeout has passed.\n\n        In the event of a timeout, an exception will be thrown. The\n        default timeout is 5 seconds; it may be overridden with a\n        ``timeout`` keyword argument or globally with the\n        ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        If ``condition`` is not ``None``, the `.IOLoop` will be restarted\n        after `stop()` until ``condition()`` returns ``True``.\n\n        .. versionchanged:: 3.1\n           Added the ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        if timeout is None:\n            timeout = get_async_test_timeout()\n\n        if not self.__stopped:\n            if timeout:\n\n                def timeout_func() -> None:\n                    try:\n                        raise self.failureException(\n                            \"Async operation timed out after %s seconds\" % timeout\n                        )\n                    except Exception:\n                        self.__failure = sys.exc_info()\n                    self.stop()\n\n                self.__timeout = self.io_loop.add_timeout(\n                    self.io_loop.time() + timeout, timeout_func\n                )\n            while True:\n                self.__running = True\n                self.io_loop.start()\n                if self.__failure is not None or condition is None or condition():\n                    break\n            if self.__timeout is not None:\n                self.io_loop.remove_timeout(self.__timeout)\n                self.__timeout = None\n        assert self.__stopped\n        self.__stopped = False\n        self.__rethrow()\n        result = self.__stop_args\n        self.__stop_args = None\n        return result",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 120,
                            "end_lineno": 393,
                            "parent": null
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                }
            ]
        }
    ],
    "old_commit_hash": "f661b2a19358e1dde23fdd7fb740c74cfa78b375^",
    "new_commit_hash": "f661b2a19358e1dde23fdd7fb740c74cfa78b375",
    "commit_message": "Re-simpify AsyncTestCase setUp & tearDown",
    "commit_date": "2023-02-10T14:28:50Z",
    "metadata": {}
}
{
    "repo_name": "tornado",
    "new_commit_hash": "f661b2a19358e1dde23fdd7fb740c74cfa78b375",
    "test_file_codes": [
        "# Licensed under the Apache License, Version 2.0 (the \"License\"); you may\n# not use this file except in compliance with the License. You may obtain\n# a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT\n# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\n# License for the specific language governing permissions and limitations\n# under the License.\n\nimport asyncio\nimport unittest\nimport warnings\n\nfrom concurrent.futures import ThreadPoolExecutor\nfrom tornado import gen\nfrom tornado.ioloop import IOLoop\nfrom tornado.platform.asyncio import (\n    AsyncIOLoop,\n    to_asyncio_future,\n    AnyThreadEventLoopPolicy,\n)\nfrom tornado.testing import AsyncTestCase, gen_test\n\n\nclass AsyncIOLoopTest(AsyncTestCase):\n    def get_new_ioloop(self):\n        io_loop = AsyncIOLoop(make_current=False)\n        return io_loop\n\n    @property\n    def asyncio_loop(self):\n        return self.io_loop.asyncio_loop\n\n    def test_asyncio_callback(self):\n        # Basic test that the asyncio loop is set up correctly.\n        async def add_callback():\n            asyncio.get_event_loop().call_soon(self.stop)\n\n        self.asyncio_loop.run_until_complete(add_callback())\n        self.wait()\n\n    @gen_test\n    def test_asyncio_future(self):\n        # Test that we can yield an asyncio future from a tornado coroutine.\n        # Without 'yield from', we must wrap coroutines in ensure_future,\n        # which was introduced during Python 3.4, deprecating the prior \"async\".\n        if hasattr(asyncio, \"ensure_future\"):\n            ensure_future = asyncio.ensure_future\n        else:\n            # async is a reserved word in Python 3.7\n            ensure_future = getattr(asyncio, \"async\")\n\n        x = yield ensure_future(\n            asyncio.get_event_loop().run_in_executor(None, lambda: 42)\n        )\n        self.assertEqual(x, 42)\n\n    @gen_test\n    def test_asyncio_yield_from(self):\n        @gen.coroutine\n        def f():\n            event_loop = asyncio.get_event_loop()\n            x = yield from event_loop.run_in_executor(None, lambda: 42)\n            return x\n\n        result = yield f()\n        self.assertEqual(result, 42)\n\n    def test_asyncio_adapter(self):\n        # This test demonstrates that when using the asyncio coroutine\n        # runner (i.e. run_until_complete), the to_asyncio_future\n        # adapter is needed. No adapter is needed in the other direction,\n        # as demonstrated by other tests in the package.\n        @gen.coroutine\n        def tornado_coroutine():\n            yield gen.moment\n            raise gen.Return(42)\n\n        async def native_coroutine_without_adapter():\n            return await tornado_coroutine()\n\n        async def native_coroutine_with_adapter():\n            return await to_asyncio_future(tornado_coroutine())\n\n        # Use the adapter, but two degrees from the tornado coroutine.\n        async def native_coroutine_with_adapter2():\n            return await to_asyncio_future(native_coroutine_without_adapter())\n\n        # Tornado supports native coroutines both with and without adapters\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_without_adapter), 42)\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_with_adapter), 42)\n        self.assertEqual(self.io_loop.run_sync(native_coroutine_with_adapter2), 42)\n\n        # Asyncio only supports coroutines that yield asyncio-compatible\n        # Futures (which our Future is since 5.0).\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_without_adapter()),\n            42,\n        )\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_with_adapter()),\n            42,\n        )\n        self.assertEqual(\n            self.asyncio_loop.run_until_complete(native_coroutine_with_adapter2()),\n            42,\n        )\n        # I'm not entirely sure why this manual cleanup is necessary but without\n        # it we have at-a-distance failures in ioloop_test.TestIOLoopCurrent.\n        asyncio.set_event_loop(None)\n\n\nclass LeakTest(unittest.TestCase):\n    def setUp(self):\n        # Trigger a cleanup of the mapping so we start with a clean slate.\n        AsyncIOLoop(make_current=False).close()\n        # If we don't clean up after ourselves other tests may fail on\n        # py34.\n        self.orig_policy = asyncio.get_event_loop_policy()\n        asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())\n\n    def tearDown(self):\n        try:\n            loop = asyncio.get_event_loop_policy().get_event_loop()\n        except Exception:\n            # We may not have a current event loop at this point.\n            pass\n        else:\n            loop.close()\n        asyncio.set_event_loop_policy(self.orig_policy)\n\n    def test_ioloop_close_leak(self):\n        orig_count = len(IOLoop._ioloop_for_asyncio)\n        for i in range(10):\n            # Create and close an AsyncIOLoop using Tornado interfaces.\n            with warnings.catch_warnings():\n                warnings.simplefilter(\"ignore\", DeprecationWarning)\n                loop = AsyncIOLoop()\n                loop.close()\n        new_count = len(IOLoop._ioloop_for_asyncio) - orig_count\n        self.assertEqual(new_count, 0)\n\n    def test_asyncio_close_leak(self):\n        orig_count = len(IOLoop._ioloop_for_asyncio)\n        for i in range(10):\n            # Create and close an AsyncIOMainLoop using asyncio interfaces.\n            loop = asyncio.new_event_loop()\n            loop.call_soon(IOLoop.current)\n            loop.call_soon(loop.stop)\n            loop.run_forever()\n            loop.close()\n        new_count = len(IOLoop._ioloop_for_asyncio) - orig_count\n        # Because the cleanup is run on new loop creation, we have one\n        # dangling entry in the map (but only one).\n        self.assertEqual(new_count, 1)\n\n\nclass AnyThreadEventLoopPolicyTest(unittest.TestCase):\n    def setUp(self):\n        self.orig_policy = asyncio.get_event_loop_policy()\n        self.executor = ThreadPoolExecutor(1)\n\n    def tearDown(self):\n        asyncio.set_event_loop_policy(self.orig_policy)\n        self.executor.shutdown()\n\n    def get_event_loop_on_thread(self):\n        def get_and_close_event_loop():\n            \"\"\"Get the event loop. Close it if one is returned.\n\n            Returns the (closed) event loop. This is a silly thing\n            to do and leaves the thread in a broken state, but it's\n            enough for this test. Closing the loop avoids resource\n            leak warnings.\n            \"\"\"\n            loop = asyncio.get_event_loop()\n            loop.close()\n            return loop\n\n        future = self.executor.submit(get_and_close_event_loop)\n        return future.result()\n\n    def run_policy_test(self, accessor, expected_type):\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            # With the default policy, non-main threads don't get an event\n            # loop.\n            self.assertRaises(\n                (RuntimeError, AssertionError), self.executor.submit(accessor).result\n            )\n            # Set the policy and we can get a loop.\n            asyncio.set_event_loop_policy(AnyThreadEventLoopPolicy())\n            self.assertIsInstance(\n                self.executor.submit(accessor).result(), expected_type\n            )\n            # Clean up to silence leak warnings. Always use asyncio since\n            # IOLoop doesn't (currently) close the underlying loop.\n            self.executor.submit(lambda: asyncio.get_event_loop().close()).result()  # type: ignore\n\n    def test_asyncio_accessor(self):\n        self.run_policy_test(asyncio.get_event_loop, asyncio.AbstractEventLoop)\n\n    def test_tornado_accessor(self):\n        self.run_policy_test(IOLoop.current, IOLoop)\n",
        "#!/usr/bin/env python\nimport unittest\nimport sys\nimport time\nimport traceback\nimport platform\nimport os\n\n\nclass PytestLikeResult(unittest.TestResult):\n    def __init__(self, *args, **kwargs):\n        super().__init__(*args, **kwargs)\n        self.start_time = None\n        # Store (test, outcome, info) for each test:\n        #   outcome ∈ { \"passed\", \"failed\", \"error\", \"skipped\" }\n        #   info can hold the exception or skip reason, etc.\n        self.results = []\n\n    def startTestRun(self):\n        \"\"\"Called once before all tests.\"\"\"\n        self.start_time = time.time()\n        # Print a header similar to pytest\n        print(\"=\" * 29 + \" test session starts \" + \"=\" * 29)\n        # You can customize the platform/versions/paths as you like:\n        print(\n            f\"platform {platform.system().lower()} -- Python {platform.python_version()}\"\n        )\n        print(f\"rootdir: {os.path.abspath(os.curdir)}\")\n\n    def stopTestRun(self):\n        \"\"\"Called once after all tests are done.\"\"\"\n        elapsed = time.time() - self.start_time\n        print()  # blank line after final dot\n\n        # Summarize\n        ntests = len(self.results)\n        print(f\"collected {ntests} items\")\n\n        # Show details of failures/errors\n        failures = [r for r in self.results if r[1] == \"failed\"]\n        errors = [r for r in self.results if r[1] == \"error\"]\n\n        if failures:\n            print(\"\\n\" + \"=\" * 35 + \" FAILURES \" + \"=\" * 35)\n            for test, outcome, err_info in failures:\n                test = test.split(\"::\")\n                if len(test) == 3:\n                    test = f\"{test[1]}.{test[2]}\"\n                else:\n                    test = f\"{test[1]}\"\n                print(f\"__________ {test} __________\")\n                self._print_traceback(err_info)\n        if errors:\n            print(\"\\n\" + \"=\" * 35 + \" ERRORS \" + \"=\" * 37)\n            for test, outcome, err_info in errors:\n                test = test.split(\"::\")\n                if len(test) == 3:\n                    test = f\"{test[1]}.{test[2]}\"\n                else:\n                    test = f\"{test[1]}\"\n                print(f\"__________ {test} __________\")\n                self._print_traceback(err_info)\n\n        # Optional short summary lines for passes/fails/skips\n        print(\"\\n==================== short test summary info ====================\")\n        for test, outcome, info in self.results:\n            if outcome in (\"passed\", \"failed\", \"error\", \"skipped\"):\n                print(f\"{outcome.upper()} {test}\")\n\n        # Final one-line summary: “1 failed, 1 errors, 10 passed in 2.04s”\n        passed_count = sum(1 for r in self.results if r[1] == \"passed\")\n        failed_count = len(failures)\n        error_count = len(errors)\n        skipped_count = sum(1 for r in self.results if r[1] == \"skipped\")\n\n        summary_bits = []\n        if failed_count:\n            summary_bits.append(f\"{failed_count} failed\")\n        if error_count:\n            summary_bits.append(f\"{error_count} error{'s' if error_count != 1 else ''}\")\n        if passed_count:\n            summary_bits.append(f\"{passed_count} passed\")\n        if skipped_count:\n            summary_bits.append(f\"{skipped_count} skipped\")\n\n        summary_str = \", \".join(summary_bits) if summary_bits else \"no tests run\"\n        print(\n            f\"=================== {summary_str} in {elapsed:.2f}s ===================\"\n        )\n\n    def startTest(self, test):\n        \"\"\"Called right before each test method.\"\"\"\n        super().startTest(test)\n        # Print a dot or some indicator\n        print(\".\", end=\"\", flush=True)\n\n    def addSuccess(self, test):\n        super().addSuccess(test)\n        self.results.append((self._test_id(test), \"passed\", None))\n\n    def addFailure(self, test, err):\n        super().addFailure(test, err)\n        self.results.append((self._test_id(test), \"failed\", err))\n\n    def addError(self, test, err):\n        super().addError(test, err)\n        self.results.append((self._test_id(test), \"error\", err))\n\n    def addSkip(self, test, reason):\n        super().addSkip(test, reason)\n        self.results.append((self._test_id(test), \"skipped\", reason))\n\n    def _print_traceback(self, err_info):\n        \"\"\"Helper to print the traceback like pytest does.\"\"\"\n        if isinstance(err_info, tuple) and len(err_info) == 3:\n            exc_type, exc_value, tb = err_info\n            traceback.print_exception(exc_type, exc_value, tb, file=sys.stdout)\n        else:\n            print(str(err_info))\n\n    def _test_id(self, test):\n        try:\n            return f\"{test.__class__.__module__}::{test.__class__.__name__}::{test._testMethodName}\"\n        except AttributeError:\n            # Fallback for _ErrorHolder objects\n            return f\"{test.__class__.__module__}::{test.__class__.__name__}\"\n\n\nclass PytestLikeRunner(unittest.TextTestRunner):\n    \"\"\"A custom TextTestRunner that uses the PytestLikeResult.\"\"\"\n\n    resultclass = PytestLikeResult\n\n    def run(self, test):\n        result = self._makeResult()\n        # Overriding run to ensure we call startTestRun/stopTestRun\n        result.startTestRun()\n        test(result)\n        result.stopTestRun()\n        return result\n\n    def tearDown(self):\n        # Reset signal handlers\n        from twisted.internet import reactor\n\n        reactor.removeAll()\n        reactor.stop()\n        super().tearDown()\n\n\ndef main():\n    loader = unittest.TestLoader()\n    suite = loader.discover(\".\", pattern=\"test_*.py\")  # discover tests in r2e_tests/\n    runner = PytestLikeRunner(verbosity=0)\n    result = runner.run(suite)\n    # Exit with code 0 if all tests passed/skipped, 1 if there were fails/errors\n    sys.exit(not result.wasSuccessful())\n\n\nif __name__ == \"__main__\":\n    main()\n"
    ],
    "test_file_names": [
        "test_1.py",
        "tornado_unittest_runner.py"
    ],
    "setup_res_code": 0,
    "setup_res_stdout": "",
    "setup_res_stderr": "Using CPython 3.9.21\nCreating virtual environment at: .venv\nActivate with: source .venv/bin/activate\nResolved 1 package in 719ms\nPrepared 1 package in 749ms\nInstalled 1 package in 0.84ms\n + tornado==6.3.dev1 (from file:///home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_f661b2a19358e1dde23fdd7fb740c74cfa78b375)\nResolved 29 packages in 13ms\nInstalled 29 packages in 103ms\n + attrs==24.3.0\n + automat==24.8.1\n + cachetools==5.5.0\n + cffi==1.17.1\n + chardet==5.2.0\n + colorama==0.4.6\n + constantly==23.10.4\n + distlib==0.3.9\n + exceptiongroup==1.2.2\n + filelock==3.16.1\n + hyperlink==21.0.0\n + idna==3.10\n + incremental==24.7.2\n + iniconfig==2.0.0\n + packaging==24.2\n + platformdirs==4.3.6\n + pluggy==1.5.0\n + pycares==4.5.0\n + pycparser==2.22\n + pycurl==7.45.4\n + pyproject-api==1.8.0\n + pytest==8.3.4\n + setuptools==75.6.0\n + tomli==2.2.1\n + tox==4.23.2\n + twisted==24.11.0\n + typing-extensions==4.12.2\n + virtualenv==20.28.0\n + zope-interface==7.2\n",
    "new_commit_res_code": 1,
    "new_commit_res_stdout": "============================= test session starts =============================\nplatform linux -- Python 3.9.21\nrootdir: /home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_f661b2a19358e1dde23fdd7fb740c74cfa78b375\n........\ncollected 8 items\n\n=================================== FAILURES ===================================\n__________ AnyThreadEventLoopPolicyTest.test_tornado_accessor __________\nTraceback (most recent call last):\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 59, in testPartExecutor\n    yield\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 592, in run\n    self._callTestMethod(testMethod)\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 550, in _callTestMethod\n    method()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_f661b2a19358e1dde23fdd7fb740c74cfa78b375/r2e_tests/test_1.py\", line 207, in test_tornado_accessor\n    self.run_policy_test(IOLoop.current, IOLoop)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_f661b2a19358e1dde23fdd7fb740c74cfa78b375/r2e_tests/test_1.py\", line 191, in run_policy_test\n    self.assertRaises(\nAssertionError: (<class 'RuntimeError'>, <class 'AssertionError'>) not raised by result\n\n==================== short test summary info ====================\nPASSED r2e_tests.test_1::AnyThreadEventLoopPolicyTest::test_asyncio_accessor\nFAILED r2e_tests.test_1::AnyThreadEventLoopPolicyTest::test_tornado_accessor\nPASSED r2e_tests.test_1::AsyncIOLoopTest::test_asyncio_adapter\nPASSED r2e_tests.test_1::AsyncIOLoopTest::test_asyncio_callback\nPASSED r2e_tests.test_1::AsyncIOLoopTest::test_asyncio_future\nPASSED r2e_tests.test_1::AsyncIOLoopTest::test_asyncio_yield_from\nPASSED r2e_tests.test_1::LeakTest::test_asyncio_close_leak\nPASSED r2e_tests.test_1::LeakTest::test_ioloop_close_leak\n=================== 1 failed, 7 passed in 0.01s ===================\n",
    "new_commit_res_stderr": "",
    "old_commit_res_code": 1,
    "old_commit_res_stdout": "============================= test session starts =============================\nplatform linux -- Python 3.9.21\nrootdir: /home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_f661b2a19358e1dde23fdd7fb740c74cfa78b375\n........\ncollected 8 items\n\n=================================== FAILURES ===================================\n__________ AnyThreadEventLoopPolicyTest.test_tornado_accessor __________\nTraceback (most recent call last):\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 59, in testPartExecutor\n    yield\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 592, in run\n    self._callTestMethod(testMethod)\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 550, in _callTestMethod\n    method()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_f661b2a19358e1dde23fdd7fb740c74cfa78b375/r2e_tests/test_1.py\", line 207, in test_tornado_accessor\n    self.run_policy_test(IOLoop.current, IOLoop)\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_f661b2a19358e1dde23fdd7fb740c74cfa78b375/r2e_tests/test_1.py\", line 191, in run_policy_test\n    self.assertRaises(\nAssertionError: (<class 'RuntimeError'>, <class 'AssertionError'>) not raised by result\n\n=================================== ERRORS =====================================\n__________ AsyncIOLoopTest.test_asyncio_adapter __________\nTraceback (most recent call last):\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_f661b2a19358e1dde23fdd7fb740c74cfa78b375/tornado/testing.py\", line 216, in setUp\n    self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()\nAttributeError: can't set attribute\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 59, in testPartExecutor\n    yield\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 588, in run\n    self._callSetUp()\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 547, in _callSetUp\n    self.setUp()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_f661b2a19358e1dde23fdd7fb740c74cfa78b375/tornado/testing.py\", line 218, in setUp\n    self.asyncio_loop = asyncio.new_event_loop()\nAttributeError: can't set attribute\n__________ AsyncIOLoopTest.test_asyncio_callback __________\nTraceback (most recent call last):\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_f661b2a19358e1dde23fdd7fb740c74cfa78b375/tornado/testing.py\", line 216, in setUp\n    self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()\nAttributeError: can't set attribute\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 59, in testPartExecutor\n    yield\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 588, in run\n    self._callSetUp()\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 547, in _callSetUp\n    self.setUp()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_f661b2a19358e1dde23fdd7fb740c74cfa78b375/tornado/testing.py\", line 218, in setUp\n    self.asyncio_loop = asyncio.new_event_loop()\nAttributeError: can't set attribute\n__________ AsyncIOLoopTest.test_asyncio_future __________\nTraceback (most recent call last):\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_f661b2a19358e1dde23fdd7fb740c74cfa78b375/tornado/testing.py\", line 216, in setUp\n    self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()\nAttributeError: can't set attribute\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 59, in testPartExecutor\n    yield\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 588, in run\n    self._callSetUp()\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 547, in _callSetUp\n    self.setUp()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_f661b2a19358e1dde23fdd7fb740c74cfa78b375/tornado/testing.py\", line 218, in setUp\n    self.asyncio_loop = asyncio.new_event_loop()\nAttributeError: can't set attribute\n__________ AsyncIOLoopTest.test_asyncio_yield_from __________\nTraceback (most recent call last):\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_f661b2a19358e1dde23fdd7fb740c74cfa78b375/tornado/testing.py\", line 216, in setUp\n    self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()\nAttributeError: can't set attribute\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 59, in testPartExecutor\n    yield\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 588, in run\n    self._callSetUp()\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 547, in _callSetUp\n    self.setUp()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_f661b2a19358e1dde23fdd7fb740c74cfa78b375/tornado/testing.py\", line 218, in setUp\n    self.asyncio_loop = asyncio.new_event_loop()\nAttributeError: can't set attribute\n\n==================== short test summary info ====================\nPASSED r2e_tests.test_1::AnyThreadEventLoopPolicyTest::test_asyncio_accessor\nFAILED r2e_tests.test_1::AnyThreadEventLoopPolicyTest::test_tornado_accessor\nERROR r2e_tests.test_1::AsyncIOLoopTest::test_asyncio_adapter\nERROR r2e_tests.test_1::AsyncIOLoopTest::test_asyncio_callback\nERROR r2e_tests.test_1::AsyncIOLoopTest::test_asyncio_future\nERROR r2e_tests.test_1::AsyncIOLoopTest::test_asyncio_yield_from\nPASSED r2e_tests.test_1::LeakTest::test_asyncio_close_leak\nPASSED r2e_tests.test_1::LeakTest::test_ioloop_close_leak\n=================== 1 failed, 4 errors, 3 passed in 0.00s ===================\n",
    "old_commit_res_stderr": ""
}
tornado/test/asyncio_test.py tornado/testing.py
We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[ISSUE]
**Title:** AttributeError When Assigning `asyncio_loop` in `AsyncTestCase` Setup

**Description:**
When using `AsyncTestCase` for asynchronous testing, the `setUp` method tries to assign a new asyncio event loop to `self.asyncio_loop` using `asyncio.new_event_loop()`. This results in an `AttributeError`, preventing the test environment from being set up properly.

**Example Code:**
```python
class MyAsyncTest(AsyncTestCase):

    def setUp(self):
        self.asyncio_loop = asyncio.new_event_loop()
        self.io_loop = self.get_new_ioloop()
        self.io_loop.make_current()
```

**Expected Behavior:**
The `setUp` method should successfully initialize the `asyncio_loop` without raising any errors, allowing asynchronous tests to proceed as intended.

**Actual Behavior:**
An `AttributeError` is raised when attempting to assign `self.asyncio_loop`, disrupting the test setup process.

**Error Message:**
```
AttributeError: can't set attribute
```
[/ISSUE]
--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.

--- BEGIN FILE ---
```
### tornado/testing.py
"""Support classes for automated testing.

* `AsyncTestCase` and `AsyncHTTPTestCase`:  Subclasses of unittest.TestCase
  with additional support for testing asynchronous (`.IOLoop`-based) code.

* `ExpectLog`: Make test logs less spammy.

* `main()`: A simple test runner (wrapper around unittest.main()) with support
  for the tornado.autoreload module to rerun the tests when code changes.
"""

import asyncio
from collections.abc import Generator
import functools
import inspect
import logging
import os
import re
import signal
import socket
import sys
import unittest
import warnings

from tornado import gen
from tornado.httpclient import AsyncHTTPClient, HTTPResponse
from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop, TimeoutError
from tornado import netutil
from tornado.platform.asyncio import AsyncIOMainLoop
from tornado.process import Subprocess
from tornado.log import app_log
from tornado.util import raise_exc_info, basestring_type
from tornado.web import Application

import typing
from typing import Tuple, Any, Callable, Type, Dict, Union, Optional, Coroutine
from types import TracebackType

if typing.TYPE_CHECKING:
    _ExcInfoTuple = Tuple[
        Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]
    ]


_NON_OWNED_IOLOOPS = AsyncIOMainLoop


def bind_unused_port(
    reuse_port: bool = False, address: str = "127.0.0.1"
) -> Tuple[socket.socket, int]:
    """Binds a server socket to an available port on localhost.

    Returns a tuple (socket, port).

    .. versionchanged:: 4.4
       Always binds to ``127.0.0.1`` without resolving the name
       ``localhost``.

    .. versionchanged:: 6.2
       Added optional ``address`` argument to
       override the default "127.0.0.1".
    """
    sock = netutil.bind_sockets(
        0, address, family=socket.AF_INET, reuse_port=reuse_port
    )[0]
    port = sock.getsockname()[1]
    return sock, port


def get_async_test_timeout() -> float:
    """Get the global timeout setting for async tests.

    Returns a float, the timeout in seconds.

    .. versionadded:: 3.1
    """
    env = os.environ.get("ASYNC_TEST_TIMEOUT")
    if env is not None:
        try:
            return float(env)
        except ValueError:
            pass
    return 5


class _TestMethodWrapper(object):
    """Wraps a test method to raise an error if it returns a value.

    This is mainly used to detect undecorated generators (if a test
    method yields it must use a decorator to consume the generator),
    but will also detect other kinds of return values (these are not
    necessarily errors, but we alert anyway since there is no good
    reason to return a value from a test).
    """

    def __init__(self, orig_method: Callable) -> None:
        self.orig_method = orig_method
        self.__wrapped__ = orig_method

    def __call__(self, *args: Any, **kwargs: Any) -> None:
        result = self.orig_method(*args, **kwargs)
        if isinstance(result, Generator) or inspect.iscoroutine(result):
            raise TypeError(
                "Generator and coroutine test methods should be"
                " decorated with tornado.testing.gen_test"
            )
        elif result is not None:
            raise ValueError("Return value from test method ignored: %r" % result)

    def __getattr__(self, name: str) -> Any:
        """Proxy all unknown attributes to the original method.

        This is important for some of the decorators in the `unittest`
        module, such as `unittest.skipIf`.
        """
        return getattr(self.orig_method, name)


class AsyncTestCase(unittest.TestCase):
    """`~unittest.TestCase` subclass for testing `.IOLoop`-based
    asynchronous code.

    The unittest framework is synchronous, so the test must be
    complete by the time the test method returns. This means that
    asynchronous code cannot be used in quite the same way as usual
    and must be adapted to fit. To write your tests with coroutines,
    decorate your test methods with `tornado.testing.gen_test` instead
    of `tornado.gen.coroutine`.

    This class also provides the (deprecated) `stop()` and `wait()`
    methods for a more manual style of testing. The test method itself
    must call ``self.wait()``, and asynchronous callbacks should call
    ``self.stop()`` to signal completion.

    By default, a new `.IOLoop` is constructed for each test and is available
    as ``self.io_loop``.  If the code being tested requires a
    global `.IOLoop`, subclasses should override `get_new_ioloop` to return it.

    The `.IOLoop`'s ``start`` and ``stop`` methods should not be
    called directly.  Instead, use `self.stop <stop>` and `self.wait
    <wait>`.  Arguments passed to ``self.stop`` are returned from
    ``self.wait``.  It is possible to have multiple ``wait``/``stop``
    cycles in the same test.

    Example::

        # This test uses coroutine style.
        class MyTestCase(AsyncTestCase):
            @tornado.testing.gen_test
            def test_http_fetch(self):
                client = AsyncHTTPClient()
                response = yield client.fetch("http://www.tornadoweb.org")
                # Test contents of response
                self.assertIn("FriendFeed", response.body)

        # This test uses argument passing between self.stop and self.wait.
        class MyTestCase2(AsyncTestCase):
            def test_http_fetch(self):
                client = AsyncHTTPClient()
                client.fetch("http://www.tornadoweb.org/", self.stop)
                response = self.wait()
                # Test contents of response
                self.assertIn("FriendFeed", response.body)
    """

    def __init__(self, methodName: str = "runTest") -> None:
        super().__init__(methodName)
        self.__stopped = False
        self.__running = False
        self.__failure = None  # type: Optional[_ExcInfoTuple]
        self.__stop_args = None  # type: Any
        self.__timeout = None  # type: Optional[object]

        # It's easy to forget the @gen_test decorator, but if you do
        # the test will silently be ignored because nothing will consume
        # the generator.  Replace the test method with a wrapper that will
        # make sure it's not an undecorated generator.
        setattr(self, methodName, _TestMethodWrapper(getattr(self, methodName)))

        # Not used in this class itself, but used by @gen_test
        self._test_generator = None  # type: Optional[Union[Generator, Coroutine]]

    def setUp(self) -> None:
        setup_with_context_manager(self, warnings.catch_warnings())
        warnings.filterwarnings(
            "ignore",
            message="There is no current event loop",
            category=DeprecationWarning,
            module=r"tornado\..*",
        )
        super().setUp()
        # NOTE: this code attempts to navigate deprecation warnings introduced
        # in Python 3.10. The idea of an implicit current event loop is
        # deprecated in that version, with the intention that tests like this
        # explicitly create a new event loop and run on it. However, other
        # packages such as pytest-asyncio (as of version 0.16.0) still rely on
        # the implicit current event loop and we want to be compatible with them
        # (even when run on 3.10, but not, of course, on the future version of
        # python that removes the get/set_event_loop methods completely).
        #
        # Deprecation warnings were introduced inconsistently:
        # asyncio.get_event_loop warns, but
        # asyncio.get_event_loop_policy().get_event_loop does not. Similarly,
        # none of the set_event_loop methods warn, although comments on
        # https://bugs.python.org/issue39529 indicate that they are also
        # intended for future removal.
        #
        # Therefore, we first attempt to access the event loop with the
        # (non-warning) policy method, and if it fails, fall back to creating a
        # new event loop. We do not have effective test coverage of the
        # new event loop case; this will have to be watched when/if
        # get_event_loop is actually removed.
        self.should_close_asyncio_loop = False
        try:
            self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()
        except Exception:
            self.asyncio_loop = asyncio.new_event_loop()
            self.should_close_asyncio_loop = True

        async def get_loop() -> IOLoop:
            return self.get_new_ioloop()

        self.io_loop = self.asyncio_loop.run_until_complete(get_loop())
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)
            self.io_loop.make_current()

    def tearDown(self) -> None:
        # Native coroutines tend to produce warnings if they're not
        # allowed to run to completion. It's difficult to ensure that
        # this always happens in tests, so cancel any tasks that are
        # still pending by the time we get here.
        asyncio_loop = self.io_loop.asyncio_loop  # type: ignore
        if hasattr(asyncio, "all_tasks"):  # py37
            tasks = asyncio.all_tasks(asyncio_loop)  # type: ignore
        else:
            tasks = asyncio.Task.all_tasks(asyncio_loop)
        # Tasks that are done may still appear here and may contain
        # non-cancellation exceptions, so filter them out.
        tasks = [t for t in tasks if not t.done()]  # type: ignore
        for t in tasks:
            t.cancel()
        # Allow the tasks to run and finalize themselves (which means
        # raising a CancelledError inside the coroutine). This may
        # just transform the "task was destroyed but it is pending"
        # warning into a "uncaught CancelledError" warning, but
        # catching CancelledErrors in coroutines that may leak is
        # simpler than ensuring that no coroutines leak.
        if tasks:
            done, pending = self.io_loop.run_sync(lambda: asyncio.wait(tasks))
            assert not pending
            # If any task failed with anything but a CancelledError, raise it.
            for f in done:
                try:
                    f.result()
                except asyncio.CancelledError:
                    pass

        # Clean up Subprocess, so it can be used again with a new ioloop.
        Subprocess.uninitialize()
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)
            self.io_loop.clear_current()
        if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):
            # Try to clean up any file descriptors left open in the ioloop.
            # This avoids leaks, especially when tests are run repeatedly
            # in the same process with autoreload (because curl does not
            # set FD_CLOEXEC on its file descriptors)
            self.io_loop.close(all_fds=True)
        if self.should_close_asyncio_loop:
            self.asyncio_loop.close()
        super().tearDown()
        # In case an exception escaped or the StackContext caught an exception
        # when there wasn't a wait() to re-raise it, do so here.
        # This is our last chance to raise an exception in a way that the
        # unittest machinery understands.
        self.__rethrow()

    def get_new_ioloop(self) -> IOLoop:
        """Returns the `.IOLoop` to use for this test.

        By default, a new `.IOLoop` is created for each test.
        Subclasses may override this method to return
        `.IOLoop.current()` if it is not appropriate to use a new
        `.IOLoop` in each tests (for example, if there are global
        singletons using the default `.IOLoop`) or if a per-test event
        loop is being provided by another system (such as
        ``pytest-asyncio``).
        """
        return IOLoop(make_current=False)

    def _handle_exception(
        self, typ: Type[Exception], value: Exception, tb: TracebackType
    ) -> bool:
        if self.__failure is None:
            self.__failure = (typ, value, tb)
        else:
            app_log.error(
                "multiple unhandled exceptions in test", exc_info=(typ, value, tb)
            )
        self.stop()
        return True

    def __rethrow(self) -> None:
        if self.__failure is not None:
            failure = self.__failure
            self.__failure = None
            raise_exc_info(failure)

    def run(
        self, result: Optional[unittest.TestResult] = None
    ) -> Optional[unittest.TestResult]:
        ret = super().run(result)
        # As a last resort, if an exception escaped super.run() and wasn't
        # re-raised in tearDown, raise it here.  This will cause the
        # unittest run to fail messily, but that's better than silently
        # ignoring an error.
        self.__rethrow()
        return ret

    def stop(self, _arg: Any = None, **kwargs: Any) -> None:
        """Stops the `.IOLoop`, causing one pending (or future) call to `wait()`
        to return.

        Keyword arguments or a single positional argument passed to `stop()` are
        saved and will be returned by `wait()`.

        .. deprecated:: 5.1

           `stop` and `wait` are deprecated; use ``@gen_test`` instead.
        """
        assert _arg is None or not kwargs
        self.__stop_args = kwargs or _arg
        if self.__running:
            self.io_loop.stop()
            self.__running = False
        self.__stopped = True

    def wait(
        self,
        condition: Optional[Callable[..., bool]] = None,
        timeout: Optional[float] = None,
    ) -> Any:
        """Runs the `.IOLoop` until stop is called or timeout has passed.

        In the event of a timeout, an exception will be thrown. The
        default timeout is 5 seconds; it may be overridden with a
        ``timeout`` keyword argument or globally with the
        ``ASYNC_TEST_TIMEOUT`` environment variable.

        If ``condition`` is not ``None``, the `.IOLoop` will be restarted
        after `stop()` until ``condition()`` returns ``True``.

        .. versionchanged:: 3.1
           Added the ``ASYNC_TEST_TIMEOUT`` environment variable.

        .. deprecated:: 5.1

           `stop` and `wait` are deprecated; use ``@gen_test`` instead.
        """
        if timeout is None:
            timeout = get_async_test_timeout()

        if not self.__stopped:
            if timeout:

                def timeout_func() -> None:
                    try:
                        raise self.failureException(
                            "Async operation timed out after %s seconds" % timeout
                        )
                    except Exception:
                        self.__failure = sys.exc_info()
                    self.stop()

                self.__timeout = self.io_loop.add_timeout(
                    self.io_loop.time() + timeout, timeout_func
                )
            while True:
                self.__running = True
                self.io_loop.start()
                if self.__failure is not None or condition is None or condition():
                    break
            if self.__timeout is not None:
                self.io_loop.remove_timeout(self.__timeout)
                self.__timeout = None
        assert self.__stopped
        self.__stopped = False
        self.__rethrow()
        result = self.__stop_args
        self.__stop_args = None
        return result


class AsyncHTTPTestCase(AsyncTestCase):
    """A test case that starts up an HTTP server.

    Subclasses must override `get_app()`, which returns the
    `tornado.web.Application` (or other `.HTTPServer` callback) to be tested.
    Tests will typically use the provided ``self.http_client`` to fetch
    URLs from this server.

    Example, assuming the "Hello, world" example from the user guide is in
    ``hello.py``::

        import hello

        class TestHelloApp(AsyncHTTPTestCase):
            def get_app(self):
                return hello.make_app()

            def test_homepage(self):
                response = self.fetch('/')
                self.assertEqual(response.code, 200)
                self.assertEqual(response.body, 'Hello, world')

    That call to ``self.fetch()`` is equivalent to ::

        self.http_client.fetch(self.get_url('/'), self.stop)
        response = self.wait()

    which illustrates how AsyncTestCase can turn an asynchronous operation,
    like ``http_client.fetch()``, into a synchronous operation. If you need
    to do other asynchronous operations in tests, you'll probably need to use
    ``stop()`` and ``wait()`` yourself.
    """

    def setUp(self) -> None:
        super().setUp()
        sock, port = bind_unused_port()
        self.__port = port

        self.http_client = self.get_http_client()
        self._app = self.get_app()
        self.http_server = self.get_http_server()
        self.http_server.add_sockets([sock])

    def get_http_client(self) -> AsyncHTTPClient:
        return AsyncHTTPClient()

    def get_http_server(self) -> HTTPServer:
        return HTTPServer(self._app, **self.get_httpserver_options())

    def get_app(self) -> Application:
        """Should be overridden by subclasses to return a
        `tornado.web.Application` or other `.HTTPServer` callback.
        """
        raise NotImplementedError()

    def fetch(
        self, path: str, raise_error: bool = False, **kwargs: Any
    ) -> HTTPResponse:
        """Convenience method to synchronously fetch a URL.

        The given path will be appended to the local server's host and
        port.  Any additional keyword arguments will be passed directly to
        `.AsyncHTTPClient.fetch` (and so could be used to pass
        ``method="POST"``, ``body="..."``, etc).

        If the path begins with http:// or https://, it will be treated as a
        full URL and will be fetched as-is.

        If ``raise_error`` is ``True``, a `tornado.httpclient.HTTPError` will
        be raised if the response code is not 200. This is the same behavior
        as the ``raise_error`` argument to `.AsyncHTTPClient.fetch`, but
        the default is ``False`` here (it's ``True`` in `.AsyncHTTPClient`)
        because tests often need to deal with non-200 response codes.

        .. versionchanged:: 5.0
           Added support for absolute URLs.

        .. versionchanged:: 5.1

           Added the ``raise_error`` argument.

        .. deprecated:: 5.1

           This method currently turns any exception into an
           `.HTTPResponse` with status code 599. In Tornado 6.0,
           errors other than `tornado.httpclient.HTTPError` will be
           passed through, and ``raise_error=False`` will only
           suppress errors that would be raised due to non-200
           response codes.

        """
        if path.lower().startswith(("http://", "https://")):
            url = path
        else:
            url = self.get_url(path)
        return self.io_loop.run_sync(
            lambda: self.http_client.fetch(url, raise_error=raise_error, **kwargs),
            timeout=get_async_test_timeout(),
        )

    def get_httpserver_options(self) -> Dict[str, Any]:
        """May be overridden by subclasses to return additional
        keyword arguments for the server.
        """
        return {}

    def get_http_port(self) -> int:
        """Returns the port used by the server.

        A new port is chosen for each test.
        """
        return self.__port

    def get_protocol(self) -> str:
        return "http"

    def get_url(self, path: str) -> str:
        """Returns an absolute url for the given path on the test server."""
        return "%s://127.0.0.1:%s%s" % (self.get_protocol(), self.get_http_port(), path)

    def tearDown(self) -> None:
        self.http_server.stop()
        self.io_loop.run_sync(
            self.http_server.close_all_connections, timeout=get_async_test_timeout()
        )
        self.http_client.close()
        del self.http_server
        del self._app
        super().tearDown()


class AsyncHTTPSTestCase(AsyncHTTPTestCase):
    """A test case that starts an HTTPS server.

    Interface is generally the same as `AsyncHTTPTestCase`.
    """

    def get_http_client(self) -> AsyncHTTPClient:
        return AsyncHTTPClient(force_instance=True, defaults=dict(validate_cert=False))

    def get_httpserver_options(self) -> Dict[str, Any]:
        return dict(ssl_options=self.get_ssl_options())

    def get_ssl_options(self) -> Dict[str, Any]:
        """May be overridden by subclasses to select SSL options.

        By default includes a self-signed testing certificate.
        """
        return AsyncHTTPSTestCase.default_ssl_options()

    @staticmethod
    def default_ssl_options() -> Dict[str, Any]:
        # Testing keys were generated with:
        # openssl req -new -keyout tornado/test/test.key \
        #                     -out tornado/test/test.crt -nodes -days 3650 -x509
        module_dir = os.path.dirname(__file__)
        return dict(
            certfile=os.path.join(module_dir, "test", "test.crt"),
            keyfile=os.path.join(module_dir, "test", "test.key"),
        )

    def get_protocol(self) -> str:
        return "https"


@typing.overload
def gen_test(
    *, timeout: Optional[float] = None
) -> Callable[[Callable[..., Union[Generator, "Coroutine"]]], Callable[..., None]]:
    pass


@typing.overload  # noqa: F811
def gen_test(func: Callable[..., Union[Generator, "Coroutine"]]) -> Callable[..., None]:
    pass


def gen_test(  # noqa: F811
    func: Optional[Callable[..., Union[Generator, "Coroutine"]]] = None,
    timeout: Optional[float] = None,
) -> Union[
    Callable[..., None],
    Callable[[Callable[..., Union[Generator, "Coroutine"]]], Callable[..., None]],
]:
    """Testing equivalent of ``@gen.coroutine``, to be applied to test methods.

    ``@gen.coroutine`` cannot be used on tests because the `.IOLoop` is not
    already running.  ``@gen_test`` should be applied to test methods
    on subclasses of `AsyncTestCase`.

    Example::

        class MyTest(AsyncHTTPTestCase):
            @gen_test
            def test_something(self):
                response = yield self.http_client.fetch(self.get_url('/'))

    By default, ``@gen_test`` times out after 5 seconds. The timeout may be
    overridden globally with the ``ASYNC_TEST_TIMEOUT`` environment variable,
    or for each test with the ``timeout`` keyword argument::

        class MyTest(AsyncHTTPTestCase):
            @gen_test(timeout=10)
            def test_something_slow(self):
                response = yield self.http_client.fetch(self.get_url('/'))

    Note that ``@gen_test`` is incompatible with `AsyncTestCase.stop`,
    `AsyncTestCase.wait`, and `AsyncHTTPTestCase.fetch`. Use ``yield
    self.http_client.fetch(self.get_url())`` as shown above instead.

    .. versionadded:: 3.1
       The ``timeout`` argument and ``ASYNC_TEST_TIMEOUT`` environment
       variable.

    .. versionchanged:: 4.0
       The wrapper now passes along ``*args, **kwargs`` so it can be used
       on functions with arguments.

    """
    if timeout is None:
        timeout = get_async_test_timeout()

    def wrap(f: Callable[..., Union[Generator, "Coroutine"]]) -> Callable[..., None]:
        # Stack up several decorators to allow us to access the generator
        # object itself.  In the innermost wrapper, we capture the generator
        # and save it in an attribute of self.  Next, we run the wrapped
        # function through @gen.coroutine.  Finally, the coroutine is
        # wrapped again to make it synchronous with run_sync.
        #
        # This is a good case study arguing for either some sort of
        # extensibility in the gen decorators or cancellation support.
        @functools.wraps(f)
        def pre_coroutine(self, *args, **kwargs):
            # type: (AsyncTestCase, *Any, **Any) -> Union[Generator, Coroutine]
            # Type comments used to avoid pypy3 bug.
            result = f(self, *args, **kwargs)
            if isinstance(result, Generator) or inspect.iscoroutine(result):
                self._test_generator = result
            else:
                self._test_generator = None
            return result

        if inspect.iscoroutinefunction(f):
            coro = pre_coroutine
        else:
            coro = gen.coroutine(pre_coroutine)  # type: ignore[assignment]

        @functools.wraps(coro)
        def post_coroutine(self, *args, **kwargs):
            # type: (AsyncTestCase, *Any, **Any) -> None
            try:
                return self.io_loop.run_sync(
                    functools.partial(coro, self, *args, **kwargs), timeout=timeout
                )
            except TimeoutError as e:
                # run_sync raises an error with an unhelpful traceback.
                # If the underlying generator is still running, we can throw the
                # exception back into it so the stack trace is replaced by the
                # point where the test is stopped. The only reason the generator
                # would not be running would be if it were cancelled, which means
                # a native coroutine, so we can rely on the cr_running attribute.
                if self._test_generator is not None and getattr(
                    self._test_generator, "cr_running", True
                ):
                    self._test_generator.throw(e)
                    # In case the test contains an overly broad except
                    # clause, we may get back here.
                # Coroutine was stopped or didn't raise a useful stack trace,
                # so re-raise the original exception which is better than nothing.
                raise

        return post_coroutine

    if func is not None:
        # Used like:
        #     @gen_test
        #     def f(self):
        #         pass
        return wrap(func)
    else:
        # Used like @gen_test(timeout=10)
        return wrap


# Without this attribute, nosetests will try to run gen_test as a test
# anywhere it is imported.
gen_test.__test__ = False  # type: ignore


class ExpectLog(logging.Filter):
    """Context manager to capture and suppress expected log output.

    Useful to make tests of error conditions less noisy, while still
    leaving unexpected log entries visible.  *Not thread safe.*

    The attribute ``logged_stack`` is set to ``True`` if any exception
    stack trace was logged.

    Usage::

        with ExpectLog('tornado.application', "Uncaught exception"):
            error_response = self.fetch("/some_page")

    .. versionchanged:: 4.3
       Added the ``logged_stack`` attribute.
    """

    def __init__(
        self,
        logger: Union[logging.Logger, basestring_type],
        regex: str,
        required: bool = True,
        level: Optional[int] = None,
    ) -> None:
        """Constructs an ExpectLog context manager.

        :param logger: Logger object (or name of logger) to watch.  Pass an
            empty string to watch the root logger.
        :param regex: Regular expression to match.  Any log entries on the
            specified logger that match this regex will be suppressed.
        :param required: If true, an exception will be raised if the end of the
            ``with`` statement is reached without matching any log entries.
        :param level: A constant from the ``logging`` module indicating the
            expected log level. If this parameter is provided, only log messages
            at this level will be considered to match. Additionally, the
            supplied ``logger`` will have its level adjusted if necessary (for
            the duration of the ``ExpectLog`` to enable the expected message.

        .. versionchanged:: 6.1
           Added the ``level`` parameter.

        .. deprecated:: 6.3
           In Tornado 7.0, only ``WARNING`` and higher logging levels will be
           matched by default. To match ``INFO`` and lower levels, the ``level``
           argument must be used. This is changing to minimize differences
           between ``tornado.testing.main`` (which enables ``INFO`` logs by
           default) and most other test runners (including those in IDEs)
           which have ``INFO`` logs disabled by default.
        """
        if isinstance(logger, basestring_type):
            logger = logging.getLogger(logger)
        self.logger = logger
        self.regex = re.compile(regex)
        self.required = required
        # matched and deprecated_level_matched are a counter for the respective event.
        self.matched = 0
        self.deprecated_level_matched = 0
        self.logged_stack = False
        self.level = level
        self.orig_level = None  # type: Optional[int]

    def filter(self, record: logging.LogRecord) -> bool:
        if record.exc_info:
            self.logged_stack = True
        message = record.getMessage()
        if self.regex.match(message):
            if self.level is None and record.levelno < logging.WARNING:
                # We're inside the logging machinery here so generating a DeprecationWarning
                # here won't be reported cleanly (if warnings-as-errors is enabled, the error
                # just gets swallowed by the logging module), and even if it were it would
                # have the wrong stack trace. Just remember this fact and report it in
                # __exit__ instead.
                self.deprecated_level_matched += 1
            if self.level is not None and record.levelno != self.level:
                app_log.warning(
                    "Got expected log message %r at unexpected level (%s vs %s)"
                    % (message, logging.getLevelName(self.level), record.levelname)
                )
                return True
            self.matched += 1
            return False
        return True

    def __enter__(self) -> "ExpectLog":
        if self.level is not None and self.level < self.logger.getEffectiveLevel():
            self.orig_level = self.logger.level
            self.logger.setLevel(self.level)
        self.logger.addFilter(self)
        return self

    def __exit__(
        self,
        typ: "Optional[Type[BaseException]]",
        value: Optional[BaseException],
        tb: Optional[TracebackType],
    ) -> None:
        if self.orig_level is not None:
            self.logger.setLevel(self.orig_level)
        self.logger.removeFilter(self)
        if not typ and self.required and not self.matched:
            raise Exception("did not get expected log message")
        if (
            not typ
            and self.required
            and (self.deprecated_level_matched >= self.matched)
        ):
            warnings.warn(
                "ExpectLog matched at INFO or below without level argument",
                DeprecationWarning,
            )


# From https://nedbatchelder.com/blog/201508/using_context_managers_in_test_setup.html
def setup_with_context_manager(testcase: unittest.TestCase, cm: Any) -> Any:
    """Use a contextmanager to setUp a test case."""
    val = cm.__enter__()
    testcase.addCleanup(cm.__exit__, None, None, None)
    return val


def main(**kwargs: Any) -> None:
    """A simple test runner.

    This test runner is essentially equivalent to `unittest.main` from
    the standard library, but adds support for Tornado-style option
    parsing and log formatting. It is *not* necessary to use this
    `main` function to run tests using `AsyncTestCase`; these tests
    are self-contained and can run with any test runner.

    The easiest way to run a test is via the command line::

        python -m tornado.testing tornado.test.web_test

    See the standard library ``unittest`` module for ways in which
    tests can be specified.

    Projects with many tests may wish to define a test script like
    ``tornado/test/runtests.py``.  This script should define a method
    ``all()`` which returns a test suite and then call
    `tornado.testing.main()`.  Note that even when a test script is
    used, the ``all()`` test suite may be overridden by naming a
    single test on the command line::

        # Runs all tests
        python -m tornado.test.runtests
        # Runs one test
        python -m tornado.test.runtests tornado.test.web_test

    Additional keyword arguments passed through to ``unittest.main()``.
    For example, use ``tornado.testing.main(verbosity=2)``
    to show many test details as they are run.
    See http://docs.python.org/library/unittest.html#unittest.main
    for full argument list.

    .. versionchanged:: 5.0

       This function produces no output of its own; only that produced
       by the `unittest` module (previously it would add a PASS or FAIL
       log message).
    """
    from tornado.options import define, options, parse_command_line

    define(
        "exception_on_interrupt",
        type=bool,
        default=True,
        help=(
            "If true (default), ctrl-c raises a KeyboardInterrupt "
            "exception.  This prints a stack trace but cannot interrupt "
            "certain operations.  If false, the process is more reliably "
            "killed, but does not print a stack trace."
        ),
    )

    # support the same options as unittest's command-line interface
    define("verbose", type=bool)
    define("quiet", type=bool)
    define("failfast", type=bool)
    define("catch", type=bool)
    define("buffer", type=bool)

    argv = [sys.argv[0]] + parse_command_line(sys.argv)

    if not options.exception_on_interrupt:
        signal.signal(signal.SIGINT, signal.SIG_DFL)

    if options.verbose is not None:
        kwargs["verbosity"] = 2
    if options.quiet is not None:
        kwargs["verbosity"] = 0
    if options.failfast is not None:
        kwargs["failfast"] = True
    if options.catch is not None:
        kwargs["catchbreak"] = True
    if options.buffer is not None:
        kwargs["buffer"] = True

    if __name__ == "__main__" and len(argv) == 1:
        print("No tests specified", file=sys.stderr)
        sys.exit(1)
    # In order to be able to run tests by their fully-qualified name
    # on the command line without importing all tests here,
    # module must be set to None.  Python 3.2's unittest.main ignores
    # defaultTest if no module is given (it tries to do its own
    # test discovery, which is incompatible with auto2to3), so don't
    # set module if we're not asking for a specific test.
    if len(argv) > 1:
        unittest.main(module=None, argv=argv, **kwargs)  # type: ignore
    else:
        unittest.main(defaultTest="all", argv=argv, **kwargs)


if __name__ == "__main__":
    main()

```
--- END FILE ---

Please first localize the bug based on the issue statement, and then generate *SEARCH/REPLACE* edits to fix the issue.

Every *SEARCH/REPLACE* edit must use this format:
1. The file path
2. The start of search block: <<<<<<< SEARCH
3. A contiguous chunk of lines to search for in the existing source code
4. The dividing line: =======
5. The lines to replace into the source code
6. The end of the replace block: >>>>>>> REPLACE

Here is an example:

```python
### mathweb/flask/app.py
<<<<<<< SEARCH
from flask import Flask
=======
import math
from flask import Flask
>>>>>>> REPLACE
```
</solution>

Please note that the *SEARCH/REPLACE* edit REQUIRES PROPER INDENTATION. If you would like to add the line '        print(x)', you must fully write that out, with all those spaces before the code!
Wrap each *SEARCH/REPLACE* edit in a code block as shown in the example above. If you have multiple *SEARCH/REPLACE* edits, use a separate code block for each one.
[ISSUE]
**Title:** AttributeError When Assigning `asyncio_loop` in `AsyncTestCase` Setup

**Description:**
When using `AsyncTestCase` for asynchronous testing, the `setUp` method tries to assign a new asyncio event loop to `self.asyncio_loop` using `asyncio.new_event_loop()`. This results in an `AttributeError`, preventing the test environment from being set up properly.

**Example Code:**
```python
class MyAsyncTest(AsyncTestCase):

    def setUp(self):
        self.asyncio_loop = asyncio.new_event_loop()
        self.io_loop = self.get_new_ioloop()
        self.io_loop.make_current()
```

**Expected Behavior:**
The `setUp` method should successfully initialize the `asyncio_loop` without raising any errors, allowing asynchronous tests to proceed as intended.

**Actual Behavior:**
An `AttributeError` is raised when attempting to assign `self.asyncio_loop`, disrupting the test setup process.

**Error Message:**
```
AttributeError: can't set attribute
```
[/ISSUE]
{
    "AnyThreadEventLoopPolicyTest.test_asyncio_accessor": "PASSED",
    "AnyThreadEventLoopPolicyTest.test_tornado_accessor": "FAILED",
    "AsyncIOLoopTest.test_asyncio_adapter": "PASSED",
    "AsyncIOLoopTest.test_asyncio_callback": "PASSED",
    "AsyncIOLoopTest.test_asyncio_future": "PASSED",
    "AsyncIOLoopTest.test_asyncio_yield_from": "PASSED",
    "LeakTest.test_asyncio_close_leak": "PASSED",
    "LeakTest.test_ioloop_close_leak": "PASSED"
}
diff --git a/tornado/testing.py b/tornado/testing.py
index 0232b219..508b1273 100644
--- a/tornado/testing.py
+++ b/tornado/testing.py
@@ -182,49 +182,9 @@ class AsyncTestCase(unittest.TestCase):
         self._test_generator = None  # type: Optional[Union[Generator, Coroutine]]
 
     def setUp(self) -> None:
-        setup_with_context_manager(self, warnings.catch_warnings())
-        warnings.filterwarnings(
-            "ignore",
-            message="There is no current event loop",
-            category=DeprecationWarning,
-            module=r"tornado\..*",
-        )
         super().setUp()
-        # NOTE: this code attempts to navigate deprecation warnings introduced
-        # in Python 3.10. The idea of an implicit current event loop is
-        # deprecated in that version, with the intention that tests like this
-        # explicitly create a new event loop and run on it. However, other
-        # packages such as pytest-asyncio (as of version 0.16.0) still rely on
-        # the implicit current event loop and we want to be compatible with them
-        # (even when run on 3.10, but not, of course, on the future version of
-        # python that removes the get/set_event_loop methods completely).
-        #
-        # Deprecation warnings were introduced inconsistently:
-        # asyncio.get_event_loop warns, but
-        # asyncio.get_event_loop_policy().get_event_loop does not. Similarly,
-        # none of the set_event_loop methods warn, although comments on
-        # https://bugs.python.org/issue39529 indicate that they are also
-        # intended for future removal.
-        #
-        # Therefore, we first attempt to access the event loop with the
-        # (non-warning) policy method, and if it fails, fall back to creating a
-        # new event loop. We do not have effective test coverage of the
-        # new event loop case; this will have to be watched when/if
-        # get_event_loop is actually removed.
-        self.should_close_asyncio_loop = False
-        try:
-            self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()
-        except Exception:
-            self.asyncio_loop = asyncio.new_event_loop()
-            self.should_close_asyncio_loop = True
-
-        async def get_loop() -> IOLoop:
-            return self.get_new_ioloop()
-
-        self.io_loop = self.asyncio_loop.run_until_complete(get_loop())
-        with warnings.catch_warnings():
-            warnings.simplefilter("ignore", DeprecationWarning)
-            self.io_loop.make_current()
+        self.io_loop = self.get_new_ioloop()
+        self.io_loop.make_current()
 
     def tearDown(self) -> None:
         # Native coroutines tend to produce warnings if they're not
@@ -259,17 +219,13 @@ class AsyncTestCase(unittest.TestCase):
 
         # Clean up Subprocess, so it can be used again with a new ioloop.
         Subprocess.uninitialize()
-        with warnings.catch_warnings():
-            warnings.simplefilter("ignore", DeprecationWarning)
-            self.io_loop.clear_current()
+        self.io_loop.clear_current()
         if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):
             # Try to clean up any file descriptors left open in the ioloop.
             # This avoids leaks, especially when tests are run repeatedly
             # in the same process with autoreload (because curl does not
             # set FD_CLOEXEC on its file descriptors)
             self.io_loop.close(all_fds=True)
-        if self.should_close_asyncio_loop:
-            self.asyncio_loop.close()
         super().tearDown()
         # In case an exception escaped or the StackContext caught an exception
         # when there wasn't a wait() to re-raise it, do so here.
 

{"tornado/testing.py": "\"\"\"Support classes for automated testing.\n\n* `AsyncTestCase` and `AsyncHTTPTestCase`:  Subclasses of unittest.TestCase\n  with additional support for testing asynchronous (`.IOLoop`-based) code.\n\n* `ExpectLog`: Make test logs less spammy.\n\n* `main()`: A simple test runner (wrapper around unittest.main()) with support\n  for the tornado.autoreload module to rerun the tests when code changes.\n\"\"\"\n\nimport asyncio\nfrom collections.abc import Generator\nimport functools\nimport inspect\nimport logging\nimport os\nimport re\nimport signal\nimport socket\nimport sys\nimport unittest\nimport warnings\n\nfrom tornado import gen\nfrom tornado.httpclient import AsyncHTTPClient, HTTPResponse\nfrom tornado.httpserver import HTTPServer\nfrom tornado.ioloop import IOLoop, TimeoutError\nfrom tornado import netutil\nfrom tornado.platform.asyncio import AsyncIOMainLoop\nfrom tornado.process import Subprocess\nfrom tornado.log import app_log\nfrom tornado.util import raise_exc_info, basestring_type\nfrom tornado.web import Application\n\nimport typing\nfrom typing import Tuple, Any, Callable, Type, Dict, Union, Optional, Coroutine\nfrom types import TracebackType\n\nif typing.TYPE_CHECKING:\n    _ExcInfoTuple = Tuple[\n        Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]\n    ]\n\n\n_NON_OWNED_IOLOOPS = AsyncIOMainLoop\n\n\ndef bind_unused_port(\n    reuse_port: bool = False, address: str = \"127.0.0.1\"\n) -> Tuple[socket.socket, int]:\n    \"\"\"Binds a server socket to an available port on localhost.\n\n    Returns a tuple (socket, port).\n\n    .. versionchanged:: 4.4\n       Always binds to ``127.0.0.1`` without resolving the name\n       ``localhost``.\n\n    .. versionchanged:: 6.2\n       Added optional ``address`` argument to\n       override the default \"127.0.0.1\".\n    \"\"\"\n    sock = netutil.bind_sockets(\n        0, address, family=socket.AF_INET, reuse_port=reuse_port\n    )[0]\n    port = sock.getsockname()[1]\n    return sock, port\n\n\ndef get_async_test_timeout() -> float:\n    \"\"\"Get the global timeout setting for async tests.\n\n    Returns a float, the timeout in seconds.\n\n    .. versionadded:: 3.1\n    \"\"\"\n    env = os.environ.get(\"ASYNC_TEST_TIMEOUT\")\n    if env is not None:\n        try:\n            return float(env)\n        except ValueError:\n            pass\n    return 5\n\n\nclass _TestMethodWrapper(object):\n    \"\"\"Wraps a test method to raise an error if it returns a value.\n\n    This is mainly used to detect undecorated generators (if a test\n    method yields it must use a decorator to consume the generator),\n    but will also detect other kinds of return values (these are not\n    necessarily errors, but we alert anyway since there is no good\n    reason to return a value from a test).\n    \"\"\"\n\n    def __init__(self, orig_method: Callable) -> None:\n        self.orig_method = orig_method\n        self.__wrapped__ = orig_method\n\n    def __call__(self, *args: Any, **kwargs: Any) -> None:\n        result = self.orig_method(*args, **kwargs)\n        if isinstance(result, Generator) or inspect.iscoroutine(result):\n            raise TypeError(\n                \"Generator and coroutine test methods should be\"\n                \" decorated with tornado.testing.gen_test\"\n            )\n        elif result is not None:\n            raise ValueError(\"Return value from test method ignored: %r\" % result)\n\n    def __getattr__(self, name: str) -> Any:\n        \"\"\"Proxy all unknown attributes to the original method.\n\n        This is important for some of the decorators in the `unittest`\n        module, such as `unittest.skipIf`.\n        \"\"\"\n        return getattr(self.orig_method, name)\n\n\nclass AsyncTestCase(unittest.TestCase):\n    \"\"\"`~unittest.TestCase` subclass for testing `.IOLoop`-based\n    asynchronous code.\n\n    The unittest framework is synchronous, so the test must be\n    complete by the time the test method returns. This means that\n    asynchronous code cannot be used in quite the same way as usual\n    and must be adapted to fit. To write your tests with coroutines,\n    decorate your test methods with `tornado.testing.gen_test` instead\n    of `tornado.gen.coroutine`.\n\n    This class also provides the (deprecated) `stop()` and `wait()`\n    methods for a more manual style of testing. The test method itself\n    must call ``self.wait()``, and asynchronous callbacks should call\n    ``self.stop()`` to signal completion.\n\n    By default, a new `.IOLoop` is constructed for each test and is available\n    as ``self.io_loop``.  If the code being tested requires a\n    global `.IOLoop`, subclasses should override `get_new_ioloop` to return it.\n\n    The `.IOLoop`'s ``start`` and ``stop`` methods should not be\n    called directly.  Instead, use `self.stop <stop>` and `self.wait\n    <wait>`.  Arguments passed to ``self.stop`` are returned from\n    ``self.wait``.  It is possible to have multiple ``wait``/``stop``\n    cycles in the same test.\n\n    Example::\n\n        # This test uses coroutine style.\n        class MyTestCase(AsyncTestCase):\n            @tornado.testing.gen_test\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                response = yield client.fetch(\"http://www.tornadoweb.org\")\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n\n        # This test uses argument passing between self.stop and self.wait.\n        class MyTestCase2(AsyncTestCase):\n            def test_http_fetch(self):\n                client = AsyncHTTPClient()\n                client.fetch(\"http://www.tornadoweb.org/\", self.stop)\n                response = self.wait()\n                # Test contents of response\n                self.assertIn(\"FriendFeed\", response.body)\n    \"\"\"\n\n    def __init__(self, methodName: str = \"runTest\") -> None:\n        super().__init__(methodName)\n        self.__stopped = False\n        self.__running = False\n        self.__failure = None  # type: Optional[_ExcInfoTuple]\n        self.__stop_args = None  # type: Any\n        self.__timeout = None  # type: Optional[object]\n\n        # It's easy to forget the @gen_test decorator, but if you do\n        # the test will silently be ignored because nothing will consume\n        # the generator.  Replace the test method with a wrapper that will\n        # make sure it's not an undecorated generator.\n        setattr(self, methodName, _TestMethodWrapper(getattr(self, methodName)))\n\n        # Not used in this class itself, but used by @gen_test\n        self._test_generator = None  # type: Optional[Union[Generator, Coroutine]]\n\n    def setUp(self) -> None:\n        setup_with_context_manager(self, warnings.catch_warnings())\n        warnings.filterwarnings(\n            \"ignore\",\n            message=\"There is no current event loop\",\n            category=DeprecationWarning,\n            module=r\"tornado\\..*\",\n        )\n        super().setUp()\n        # NOTE: this code attempts to navigate deprecation warnings introduced\n        # in Python 3.10. The idea of an implicit current event loop is\n        # deprecated in that version, with the intention that tests like this\n        # explicitly create a new event loop and run on it. However, other\n        # packages such as pytest-asyncio (as of version 0.16.0) still rely on\n        # the implicit current event loop and we want to be compatible with them\n        # (even when run on 3.10, but not, of course, on the future version of\n        # python that removes the get/set_event_loop methods completely).\n        #\n        # Deprecation warnings were introduced inconsistently:\n        # asyncio.get_event_loop warns, but\n        # asyncio.get_event_loop_policy().get_event_loop does not. Similarly,\n        # none of the set_event_loop methods warn, although comments on\n        # https://bugs.python.org/issue39529 indicate that they are also\n        # intended for future removal.\n        #\n        # Therefore, we first attempt to access the event loop with the\n        # (non-warning) policy method, and if it fails, fall back to creating a\n        # new event loop. We do not have effective test coverage of the\n        # new event loop case; this will have to be watched when/if\n        # get_event_loop is actually removed.\n        self.should_close_asyncio_loop = False\n        try:\n            self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()\n        except Exception:\n            self.asyncio_loop = asyncio.new_event_loop()\n            self.should_close_asyncio_loop = True\n\n        async def get_loop() -> IOLoop:\n            return self.get_new_ioloop()\n\n        self.io_loop = self.asyncio_loop.run_until_complete(get_loop())\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.make_current()\n\n    def tearDown(self) -> None:\n        # Native coroutines tend to produce warnings if they're not\n        # allowed to run to completion. It's difficult to ensure that\n        # this always happens in tests, so cancel any tasks that are\n        # still pending by the time we get here.\n        asyncio_loop = self.io_loop.asyncio_loop  # type: ignore\n        if hasattr(asyncio, \"all_tasks\"):  # py37\n            tasks = asyncio.all_tasks(asyncio_loop)  # type: ignore\n        else:\n            tasks = asyncio.Task.all_tasks(asyncio_loop)\n        # Tasks that are done may still appear here and may contain\n        # non-cancellation exceptions, so filter them out.\n        tasks = [t for t in tasks if not t.done()]  # type: ignore\n        for t in tasks:\n            t.cancel()\n        # Allow the tasks to run and finalize themselves (which means\n        # raising a CancelledError inside the coroutine). This may\n        # just transform the \"task was destroyed but it is pending\"\n        # warning into a \"uncaught CancelledError\" warning, but\n        # catching CancelledErrors in coroutines that may leak is\n        # simpler than ensuring that no coroutines leak.\n        if tasks:\n            done, pending = self.io_loop.run_sync(lambda: asyncio.wait(tasks))\n            assert not pending\n            # If any task failed with anything but a CancelledError, raise it.\n            for f in done:\n                try:\n                    f.result()\n                except asyncio.CancelledError:\n                    pass\n\n        # Clean up Subprocess, so it can be used again with a new ioloop.\n        Subprocess.uninitialize()\n        with warnings.catch_warnings():\n            warnings.simplefilter(\"ignore\", DeprecationWarning)\n            self.io_loop.clear_current()\n        if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):\n            # Try to clean up any file descriptors left open in the ioloop.\n            # This avoids leaks, especially when tests are run repeatedly\n            # in the same process with autoreload (because curl does not\n            # set FD_CLOEXEC on its file descriptors)\n            self.io_loop.close(all_fds=True)\n        if self.should_close_asyncio_loop:\n            self.asyncio_loop.close()\n        super().tearDown()\n        # In case an exception escaped or the StackContext caught an exception\n        # when there wasn't a wait() to re-raise it, do so here.\n        # This is our last chance to raise an exception in a way that the\n        # unittest machinery understands.\n        self.__rethrow()\n\n    def get_new_ioloop(self) -> IOLoop:\n        \"\"\"Returns the `.IOLoop` to use for this test.\n\n        By default, a new `.IOLoop` is created for each test.\n        Subclasses may override this method to return\n        `.IOLoop.current()` if it is not appropriate to use a new\n        `.IOLoop` in each tests (for example, if there are global\n        singletons using the default `.IOLoop`) or if a per-test event\n        loop is being provided by another system (such as\n        ``pytest-asyncio``).\n        \"\"\"\n        return IOLoop(make_current=False)\n\n    def _handle_exception(\n        self, typ: Type[Exception], value: Exception, tb: TracebackType\n    ) -> bool:\n        if self.__failure is None:\n            self.__failure = (typ, value, tb)\n        else:\n            app_log.error(\n                \"multiple unhandled exceptions in test\", exc_info=(typ, value, tb)\n            )\n        self.stop()\n        return True\n\n    def __rethrow(self) -> None:\n        if self.__failure is not None:\n            failure = self.__failure\n            self.__failure = None\n            raise_exc_info(failure)\n\n    def run(\n        self, result: Optional[unittest.TestResult] = None\n    ) -> Optional[unittest.TestResult]:\n        ret = super().run(result)\n        # As a last resort, if an exception escaped super.run() and wasn't\n        # re-raised in tearDown, raise it here.  This will cause the\n        # unittest run to fail messily, but that's better than silently\n        # ignoring an error.\n        self.__rethrow()\n        return ret\n\n    def stop(self, _arg: Any = None, **kwargs: Any) -> None:\n        \"\"\"Stops the `.IOLoop`, causing one pending (or future) call to `wait()`\n        to return.\n\n        Keyword arguments or a single positional argument passed to `stop()` are\n        saved and will be returned by `wait()`.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        assert _arg is None or not kwargs\n        self.__stop_args = kwargs or _arg\n        if self.__running:\n            self.io_loop.stop()\n            self.__running = False\n        self.__stopped = True\n\n    def wait(\n        self,\n        condition: Optional[Callable[..., bool]] = None,\n        timeout: Optional[float] = None,\n    ) -> Any:\n        \"\"\"Runs the `.IOLoop` until stop is called or timeout has passed.\n\n        In the event of a timeout, an exception will be thrown. The\n        default timeout is 5 seconds; it may be overridden with a\n        ``timeout`` keyword argument or globally with the\n        ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        If ``condition`` is not ``None``, the `.IOLoop` will be restarted\n        after `stop()` until ``condition()`` returns ``True``.\n\n        .. versionchanged:: 3.1\n           Added the ``ASYNC_TEST_TIMEOUT`` environment variable.\n\n        .. deprecated:: 5.1\n\n           `stop` and `wait` are deprecated; use ``@gen_test`` instead.\n        \"\"\"\n        if timeout is None:\n            timeout = get_async_test_timeout()\n\n        if not self.__stopped:\n            if timeout:\n\n                def timeout_func() -> None:\n                    try:\n                        raise self.failureException(\n                            \"Async operation timed out after %s seconds\" % timeout\n                        )\n                    except Exception:\n                        self.__failure = sys.exc_info()\n                    self.stop()\n\n                self.__timeout = self.io_loop.add_timeout(\n                    self.io_loop.time() + timeout, timeout_func\n                )\n            while True:\n                self.__running = True\n                self.io_loop.start()\n                if self.__failure is not None or condition is None or condition():\n                    break\n            if self.__timeout is not None:\n                self.io_loop.remove_timeout(self.__timeout)\n                self.__timeout = None\n        assert self.__stopped\n        self.__stopped = False\n        self.__rethrow()\n        result = self.__stop_args\n        self.__stop_args = None\n        return result\n\n\nclass AsyncHTTPTestCase(AsyncTestCase):\n    \"\"\"A test case that starts up an HTTP server.\n\n    Subclasses must override `get_app()`, which returns the\n    `tornado.web.Application` (or other `.HTTPServer` callback) to be tested.\n    Tests will typically use the provided ``self.http_client`` to fetch\n    URLs from this server.\n\n    Example, assuming the \"Hello, world\" example from the user guide is in\n    ``hello.py``::\n\n        import hello\n\n        class TestHelloApp(AsyncHTTPTestCase):\n            def get_app(self):\n                return hello.make_app()\n\n            def test_homepage(self):\n                response = self.fetch('/')\n                self.assertEqual(response.code, 200)\n                self.assertEqual(response.body, 'Hello, world')\n\n    That call to ``self.fetch()`` is equivalent to ::\n\n        self.http_client.fetch(self.get_url('/'), self.stop)\n        response = self.wait()\n\n    which illustrates how AsyncTestCase can turn an asynchronous operation,\n    like ``http_client.fetch()``, into a synchronous operation. If you need\n    to do other asynchronous operations in tests, you'll probably need to use\n    ``stop()`` and ``wait()`` yourself.\n    \"\"\"\n\n    def setUp(self) -> None:\n        super().setUp()\n        sock, port = bind_unused_port()\n        self.__port = port\n\n        self.http_client = self.get_http_client()\n        self._app = self.get_app()\n        self.http_server = self.get_http_server()\n        self.http_server.add_sockets([sock])\n\n    def get_http_client(self) -> AsyncHTTPClient:\n        return AsyncHTTPClient()\n\n    def get_http_server(self) -> HTTPServer:\n        return HTTPServer(self._app, **self.get_httpserver_options())\n\n    def get_app(self) -> Application:\n        \"\"\"Should be overridden by subclasses to return a\n        `tornado.web.Application` or other `.HTTPServer` callback.\n        \"\"\"\n        raise NotImplementedError()\n\n    def fetch(\n        self, path: str, raise_error: bool = False, **kwargs: Any\n    ) -> HTTPResponse:\n        \"\"\"Convenience method to synchronously fetch a URL.\n\n        The given path will be appended to the local server's host and\n        port.  Any additional keyword arguments will be passed directly to\n        `.AsyncHTTPClient.fetch` (and so could be used to pass\n        ``method=\"POST\"``, ``body=\"...\"``, etc).\n\n        If the path begins with http:// or https://, it will be treated as a\n        full URL and will be fetched as-is.\n\n        If ``raise_error`` is ``True``, a `tornado.httpclient.HTTPError` will\n        be raised if the response code is not 200. This is the same behavior\n        as the ``raise_error`` argument to `.AsyncHTTPClient.fetch`, but\n        the default is ``False`` here (it's ``True`` in `.AsyncHTTPClient`)\n        because tests often need to deal with non-200 response codes.\n\n        .. versionchanged:: 5.0\n           Added support for absolute URLs.\n\n        .. versionchanged:: 5.1\n\n           Added the ``raise_error`` argument.\n\n        .. deprecated:: 5.1\n\n           This method currently turns any exception into an\n           `.HTTPResponse` with status code 599. In Tornado 6.0,\n           errors other than `tornado.httpclient.HTTPError` will be\n           passed through, and ``raise_error=False`` will only\n           suppress errors that would be raised due to non-200\n           response codes.\n\n        \"\"\"\n        if path.lower().startswith((\"http://\", \"https://\")):\n            url = path\n        else:\n            url = self.get_url(path)\n        return self.io_loop.run_sync(\n            lambda: self.http_client.fetch(url, raise_error=raise_error, **kwargs),\n            timeout=get_async_test_timeout(),\n        )\n\n    def get_httpserver_options(self) -> Dict[str, Any]:\n        \"\"\"May be overridden by subclasses to return additional\n        keyword arguments for the server.\n        \"\"\"\n        return {}\n\n    def get_http_port(self) -> int:\n        \"\"\"Returns the port used by the server.\n\n        A new port is chosen for each test.\n        \"\"\"\n        return self.__port\n\n    def get_protocol(self) -> str:\n        return \"http\"\n\n    def get_url(self, path: str) -> str:\n        \"\"\"Returns an absolute url for the given path on the test server.\"\"\"\n        return \"%s://127.0.0.1:%s%s\" % (self.get_protocol(), self.get_http_port(), path)\n\n    def tearDown(self) -> None:\n        self.http_server.stop()\n        self.io_loop.run_sync(\n            self.http_server.close_all_connections, timeout=get_async_test_timeout()\n        )\n        self.http_client.close()\n        del self.http_server\n        del self._app\n        super().tearDown()\n\n\nclass AsyncHTTPSTestCase(AsyncHTTPTestCase):\n    \"\"\"A test case that starts an HTTPS server.\n\n    Interface is generally the same as `AsyncHTTPTestCase`.\n    \"\"\"\n\n    def get_http_client(self) -> AsyncHTTPClient:\n        return AsyncHTTPClient(force_instance=True, defaults=dict(validate_cert=False))\n\n    def get_httpserver_options(self) -> Dict[str, Any]:\n        return dict(ssl_options=self.get_ssl_options())\n\n    def get_ssl_options(self) -> Dict[str, Any]:\n        \"\"\"May be overridden by subclasses to select SSL options.\n\n        By default includes a self-signed testing certificate.\n        \"\"\"\n        return AsyncHTTPSTestCase.default_ssl_options()\n\n    @staticmethod\n    def default_ssl_options() -> Dict[str, Any]:\n        # Testing keys were generated with:\n        # openssl req -new -keyout tornado/test/test.key \\\n        #                     -out tornado/test/test.crt -nodes -days 3650 -x509\n        module_dir = os.path.dirname(__file__)\n        return dict(\n            certfile=os.path.join(module_dir, \"test\", \"test.crt\"),\n            keyfile=os.path.join(module_dir, \"test\", \"test.key\"),\n        )\n\n    def get_protocol(self) -> str:\n        return \"https\"\n\n\n@typing.overload\ndef gen_test(\n    *, timeout: Optional[float] = None\n) -> Callable[[Callable[..., Union[Generator, \"Coroutine\"]]], Callable[..., None]]:\n    pass\n\n\n@typing.overload  # noqa: F811\ndef gen_test(func: Callable[..., Union[Generator, \"Coroutine\"]]) -> Callable[..., None]:\n    pass\n\n\ndef gen_test(  # noqa: F811\n    func: Optional[Callable[..., Union[Generator, \"Coroutine\"]]] = None,\n    timeout: Optional[float] = None,\n) -> Union[\n    Callable[..., None],\n    Callable[[Callable[..., Union[Generator, \"Coroutine\"]]], Callable[..., None]],\n]:\n    \"\"\"Testing equivalent of ``@gen.coroutine``, to be applied to test methods.\n\n    ``@gen.coroutine`` cannot be used on tests because the `.IOLoop` is not\n    already running.  ``@gen_test`` should be applied to test methods\n    on subclasses of `AsyncTestCase`.\n\n    Example::\n\n        class MyTest(AsyncHTTPTestCase):\n            @gen_test\n            def test_something(self):\n                response = yield self.http_client.fetch(self.get_url('/'))\n\n    By default, ``@gen_test`` times out after 5 seconds. The timeout may be\n    overridden globally with the ``ASYNC_TEST_TIMEOUT`` environment variable,\n    or for each test with the ``timeout`` keyword argument::\n\n        class MyTest(AsyncHTTPTestCase):\n            @gen_test(timeout=10)\n            def test_something_slow(self):\n                response = yield self.http_client.fetch(self.get_url('/'))\n\n    Note that ``@gen_test`` is incompatible with `AsyncTestCase.stop`,\n    `AsyncTestCase.wait`, and `AsyncHTTPTestCase.fetch`. Use ``yield\n    self.http_client.fetch(self.get_url())`` as shown above instead.\n\n    .. versionadded:: 3.1\n       The ``timeout`` argument and ``ASYNC_TEST_TIMEOUT`` environment\n       variable.\n\n    .. versionchanged:: 4.0\n       The wrapper now passes along ``*args, **kwargs`` so it can be used\n       on functions with arguments.\n\n    \"\"\"\n    if timeout is None:\n        timeout = get_async_test_timeout()\n\n    def wrap(f: Callable[..., Union[Generator, \"Coroutine\"]]) -> Callable[..., None]:\n        # Stack up several decorators to allow us to access the generator\n        # object itself.  In the innermost wrapper, we capture the generator\n        # and save it in an attribute of self.  Next, we run the wrapped\n        # function through @gen.coroutine.  Finally, the coroutine is\n        # wrapped again to make it synchronous with run_sync.\n        #\n        # This is a good case study arguing for either some sort of\n        # extensibility in the gen decorators or cancellation support.\n        @functools.wraps(f)\n        def pre_coroutine(self, *args, **kwargs):\n            # type: (AsyncTestCase, *Any, **Any) -> Union[Generator, Coroutine]\n            # Type comments used to avoid pypy3 bug.\n            result = f(self, *args, **kwargs)\n            if isinstance(result, Generator) or inspect.iscoroutine(result):\n                self._test_generator = result\n            else:\n                self._test_generator = None\n            return result\n\n        if inspect.iscoroutinefunction(f):\n            coro = pre_coroutine\n        else:\n            coro = gen.coroutine(pre_coroutine)  # type: ignore[assignment]\n\n        @functools.wraps(coro)\n        def post_coroutine(self, *args, **kwargs):\n            # type: (AsyncTestCase, *Any, **Any) -> None\n            try:\n                return self.io_loop.run_sync(\n                    functools.partial(coro, self, *args, **kwargs), timeout=timeout\n                )\n            except TimeoutError as e:\n                # run_sync raises an error with an unhelpful traceback.\n                # If the underlying generator is still running, we can throw the\n                # exception back into it so the stack trace is replaced by the\n                # point where the test is stopped. The only reason the generator\n                # would not be running would be if it were cancelled, which means\n                # a native coroutine, so we can rely on the cr_running attribute.\n                if self._test_generator is not None and getattr(\n                    self._test_generator, \"cr_running\", True\n                ):\n                    self._test_generator.throw(e)\n                    # In case the test contains an overly broad except\n                    # clause, we may get back here.\n                # Coroutine was stopped or didn't raise a useful stack trace,\n                # so re-raise the original exception which is better than nothing.\n                raise\n\n        return post_coroutine\n\n    if func is not None:\n        # Used like:\n        #     @gen_test\n        #     def f(self):\n        #         pass\n        return wrap(func)\n    else:\n        # Used like @gen_test(timeout=10)\n        return wrap\n\n\n# Without this attribute, nosetests will try to run gen_test as a test\n# anywhere it is imported.\ngen_test.__test__ = False  # type: ignore\n\n\nclass ExpectLog(logging.Filter):\n    \"\"\"Context manager to capture and suppress expected log output.\n\n    Useful to make tests of error conditions less noisy, while still\n    leaving unexpected log entries visible.  *Not thread safe.*\n\n    The attribute ``logged_stack`` is set to ``True`` if any exception\n    stack trace was logged.\n\n    Usage::\n\n        with ExpectLog('tornado.application', \"Uncaught exception\"):\n            error_response = self.fetch(\"/some_page\")\n\n    .. versionchanged:: 4.3\n       Added the ``logged_stack`` attribute.\n    \"\"\"\n\n    def __init__(\n        self,\n        logger: Union[logging.Logger, basestring_type],\n        regex: str,\n        required: bool = True,\n        level: Optional[int] = None,\n    ) -> None:\n        \"\"\"Constructs an ExpectLog context manager.\n\n        :param logger: Logger object (or name of logger) to watch.  Pass an\n            empty string to watch the root logger.\n        :param regex: Regular expression to match.  Any log entries on the\n            specified logger that match this regex will be suppressed.\n        :param required: If true, an exception will be raised if the end of the\n            ``with`` statement is reached without matching any log entries.\n        :param level: A constant from the ``logging`` module indicating the\n            expected log level. If this parameter is provided, only log messages\n            at this level will be considered to match. Additionally, the\n            supplied ``logger`` will have its level adjusted if necessary (for\n            the duration of the ``ExpectLog`` to enable the expected message.\n\n        .. versionchanged:: 6.1\n           Added the ``level`` parameter.\n\n        .. deprecated:: 6.3\n           In Tornado 7.0, only ``WARNING`` and higher logging levels will be\n           matched by default. To match ``INFO`` and lower levels, the ``level``\n           argument must be used. This is changing to minimize differences\n           between ``tornado.testing.main`` (which enables ``INFO`` logs by\n           default) and most other test runners (including those in IDEs)\n           which have ``INFO`` logs disabled by default.\n        \"\"\"\n        if isinstance(logger, basestring_type):\n            logger = logging.getLogger(logger)\n        self.logger = logger\n        self.regex = re.compile(regex)\n        self.required = required\n        # matched and deprecated_level_matched are a counter for the respective event.\n        self.matched = 0\n        self.deprecated_level_matched = 0\n        self.logged_stack = False\n        self.level = level\n        self.orig_level = None  # type: Optional[int]\n\n    def filter(self, record: logging.LogRecord) -> bool:\n        if record.exc_info:\n            self.logged_stack = True\n        message = record.getMessage()\n        if self.regex.match(message):\n            if self.level is None and record.levelno < logging.WARNING:\n                # We're inside the logging machinery here so generating a DeprecationWarning\n                # here won't be reported cleanly (if warnings-as-errors is enabled, the error\n                # just gets swallowed by the logging module), and even if it were it would\n                # have the wrong stack trace. Just remember this fact and report it in\n                # __exit__ instead.\n                self.deprecated_level_matched += 1\n            if self.level is not None and record.levelno != self.level:\n                app_log.warning(\n                    \"Got expected log message %r at unexpected level (%s vs %s)\"\n                    % (message, logging.getLevelName(self.level), record.levelname)\n                )\n                return True\n            self.matched += 1\n            return False\n        return True\n\n    def __enter__(self) -> \"ExpectLog\":\n        if self.level is not None and self.level < self.logger.getEffectiveLevel():\n            self.orig_level = self.logger.level\n            self.logger.setLevel(self.level)\n        self.logger.addFilter(self)\n        return self\n\n    def __exit__(\n        self,\n        typ: \"Optional[Type[BaseException]]\",\n        value: Optional[BaseException],\n        tb: Optional[TracebackType],\n    ) -> None:\n        if self.orig_level is not None:\n            self.logger.setLevel(self.orig_level)\n        self.logger.removeFilter(self)\n        if not typ and self.required and not self.matched:\n            raise Exception(\"did not get expected log message\")\n        if (\n            not typ\n            and self.required\n            and (self.deprecated_level_matched >= self.matched)\n        ):\n            warnings.warn(\n                \"ExpectLog matched at INFO or below without level argument\",\n                DeprecationWarning,\n            )\n\n\n# From https://nedbatchelder.com/blog/201508/using_context_managers_in_test_setup.html\ndef setup_with_context_manager(testcase: unittest.TestCase, cm: Any) -> Any:\n    \"\"\"Use a contextmanager to setUp a test case.\"\"\"\n    val = cm.__enter__()\n    testcase.addCleanup(cm.__exit__, None, None, None)\n    return val\n\n\ndef main(**kwargs: Any) -> None:\n    \"\"\"A simple test runner.\n\n    This test runner is essentially equivalent to `unittest.main` from\n    the standard library, but adds support for Tornado-style option\n    parsing and log formatting. It is *not* necessary to use this\n    `main` function to run tests using `AsyncTestCase`; these tests\n    are self-contained and can run with any test runner.\n\n    The easiest way to run a test is via the command line::\n\n        python -m tornado.testing tornado.test.web_test\n\n    See the standard library ``unittest`` module for ways in which\n    tests can be specified.\n\n    Projects with many tests may wish to define a test script like\n    ``tornado/test/runtests.py``.  This script should define a method\n    ``all()`` which returns a test suite and then call\n    `tornado.testing.main()`.  Note that even when a test script is\n    used, the ``all()`` test suite may be overridden by naming a\n    single test on the command line::\n\n        # Runs all tests\n        python -m tornado.test.runtests\n        # Runs one test\n        python -m tornado.test.runtests tornado.test.web_test\n\n    Additional keyword arguments passed through to ``unittest.main()``.\n    For example, use ``tornado.testing.main(verbosity=2)``\n    to show many test details as they are run.\n    See http://docs.python.org/library/unittest.html#unittest.main\n    for full argument list.\n\n    .. versionchanged:: 5.0\n\n       This function produces no output of its own; only that produced\n       by the `unittest` module (previously it would add a PASS or FAIL\n       log message).\n    \"\"\"\n    from tornado.options import define, options, parse_command_line\n\n    define(\n        \"exception_on_interrupt\",\n        type=bool,\n        default=True,\n        help=(\n            \"If true (default), ctrl-c raises a KeyboardInterrupt \"\n            \"exception.  This prints a stack trace but cannot interrupt \"\n            \"certain operations.  If false, the process is more reliably \"\n            \"killed, but does not print a stack trace.\"\n        ),\n    )\n\n    # support the same options as unittest's command-line interface\n    define(\"verbose\", type=bool)\n    define(\"quiet\", type=bool)\n    define(\"failfast\", type=bool)\n    define(\"catch\", type=bool)\n    define(\"buffer\", type=bool)\n\n    argv = [sys.argv[0]] + parse_command_line(sys.argv)\n\n    if not options.exception_on_interrupt:\n        signal.signal(signal.SIGINT, signal.SIG_DFL)\n\n    if options.verbose is not None:\n        kwargs[\"verbosity\"] = 2\n    if options.quiet is not None:\n        kwargs[\"verbosity\"] = 0\n    if options.failfast is not None:\n        kwargs[\"failfast\"] = True\n    if options.catch is not None:\n        kwargs[\"catchbreak\"] = True\n    if options.buffer is not None:\n        kwargs[\"buffer\"] = True\n\n    if __name__ == \"__main__\" and len(argv) == 1:\n        print(\"No tests specified\", file=sys.stderr)\n        sys.exit(1)\n    # In order to be able to run tests by their fully-qualified name\n    # on the command line without importing all tests here,\n    # module must be set to None.  Python 3.2's unittest.main ignores\n    # defaultTest if no module is given (it tries to do its own\n    # test discovery, which is incompatible with auto2to3), so don't\n    # set module if we're not asking for a specific test.\n    if len(argv) > 1:\n        unittest.main(module=None, argv=argv, **kwargs)  # type: ignore\n    else:\n        unittest.main(defaultTest=\"all\", argv=argv, **kwargs)\n\n\nif __name__ == \"__main__\":\n    main()\n"}
A user will ask you to solve a task. You should first draft your thinking process (inner monologue). Then, generate the solution.

Your response format must follow the template below:
<think>
Your thoughts or/and draft, like working through an exercise on scratch paper. Be as casual and as long as you want until you are confident to generate a correct solution.
</think>
<solution>
Final solution presented to the user.
</solution>
system
We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[ISSUE]
**Title:** AttributeError When Assigning `asyncio_loop` in `AsyncTestCase` Setup

**Description:**
When using `AsyncTestCase` for asynchronous testing, the `setUp` method tries to assign a new asyncio event loop to `self.asyncio_loop` using `asyncio.new_event_loop()`. This results in an `AttributeError`, preventing the test environment from being set up properly.

**Example Code:**
```python
class MyAsyncTest(AsyncTestCase):

    def setUp(self):
        self.asyncio_loop = asyncio.new_event_loop()
        self.io_loop = self.get_new_ioloop()
        self.io_loop.make_current()
```

**Expected Behavior:**
The `setUp` method should successfully initialize the `asyncio_loop` without raising any errors, allowing asynchronous tests to proceed as intended.

**Actual Behavior:**
An `AttributeError` is raised when attempting to assign `self.asyncio_loop`, disrupting the test setup process.

**Error Message:**
```
AttributeError: can't set attribute
```
[/ISSUE]
--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.

--- BEGIN FILE ---
```
### tornado/testing.py
"""Support classes for automated testing.

* `AsyncTestCase` and `AsyncHTTPTestCase`:  Subclasses of unittest.TestCase
  with additional support for testing asynchronous (`.IOLoop`-based) code.

* `ExpectLog`: Make test logs less spammy.

* `main()`: A simple test runner (wrapper around unittest.main()) with support
  for the tornado.autoreload module to rerun the tests when code changes.
"""

import asyncio
from collections.abc import Generator
import functools
import inspect
import logging
import os
import re
import signal
import socket
import sys
import unittest
import warnings

from tornado import gen
from tornado.httpclient import AsyncHTTPClient, HTTPResponse
from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop, TimeoutError
from tornado import netutil
from tornado.platform.asyncio import AsyncIOMainLoop
from tornado.process import Subprocess
from tornado.log import app_log
from tornado.util import raise_exc_info, basestring_type
from tornado.web import Application

import typing
from typing import Tuple, Any, Callable, Type, Dict, Union, Optional, Coroutine
from types import TracebackType

if typing.TYPE_CHECKING:
    _ExcInfoTuple = Tuple[
        Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]
    ]


_NON_OWNED_IOLOOPS = AsyncIOMainLoop


def bind_unused_port(
    reuse_port: bool = False, address: str = "127.0.0.1"
) -> Tuple[socket.socket, int]:
    """Binds a server socket to an available port on localhost.

    Returns a tuple (socket, port).

    .. versionchanged:: 4.4
       Always binds to ``127.0.0.1`` without resolving the name
       ``localhost``.

    .. versionchanged:: 6.2
       Added optional ``address`` argument to
       override the default "127.0.0.1".
    """
    sock = netutil.bind_sockets(
        0, address, family=socket.AF_INET, reuse_port=reuse_port
    )[0]
    port = sock.getsockname()[1]
    return sock, port


def get_async_test_timeout() -> float:
    """Get the global timeout setting for async tests.

    Returns a float, the timeout in seconds.

    .. versionadded:: 3.1
    """
    env = os.environ.get("ASYNC_TEST_TIMEOUT")
    if env is not None:
        try:
            return float(env)
        except ValueError:
            pass
    return 5


class _TestMethodWrapper(object):
    """Wraps a test method to raise an error if it returns a value.

    This is mainly used to detect undecorated generators (if a test
    method yields it must use a decorator to consume the generator),
    but will also detect other kinds of return values (these are not
    necessarily errors, but we alert anyway since there is no good
    reason to return a value from a test).
    """

    def __init__(self, orig_method: Callable) -> None:
        self.orig_method = orig_method
        self.__wrapped__ = orig_method

    def __call__(self, *args: Any, **kwargs: Any) -> None:
        result = self.orig_method(*args, **kwargs)
        if isinstance(result, Generator) or inspect.iscoroutine(result):
            raise TypeError(
                "Generator and coroutine test methods should be"
                " decorated with tornado.testing.gen_test"
            )
        elif result is not None:
            raise ValueError("Return value from test method ignored: %r" % result)

    def __getattr__(self, name: str) -> Any:
        """Proxy all unknown attributes to the original method.

        This is important for some of the decorators in the `unittest`
        module, such as `unittest.skipIf`.
        """
        return getattr(self.orig_method, name)


class AsyncTestCase(unittest.TestCase):
    """`~unittest.TestCase` subclass for testing `.IOLoop`-based
    asynchronous code.

    The unittest framework is synchronous, so the test must be
    complete by the time the test method returns. This means that
    asynchronous code cannot be used in quite the same way as usual
    and must be adapted to fit. To write your tests with coroutines,
    decorate your test methods with `tornado.testing.gen_test` instead
    of `tornado.gen.coroutine`.

    This class also provides the (deprecated) `stop()` and `wait()`
    methods for a more manual style of testing. The test method itself
    must call ``self.wait()``, and asynchronous callbacks should call
    ``self.stop()`` to signal completion.

    By default, a new `.IOLoop` is constructed for each test and is available
    as ``self.io_loop``.  If the code being tested requires a
    global `.IOLoop`, subclasses should override `get_new_ioloop` to return it.

    The `.IOLoop`'s ``start`` and ``stop`` methods should not be
    called directly.  Instead, use `self.stop <stop>` and `self.wait
    <wait>`.  Arguments passed to ``self.stop`` are returned from
    ``self.wait``.  It is possible to have multiple ``wait``/``stop``
    cycles in the same test.

    Example::

        # This test uses coroutine style.
        class MyTestCase(AsyncTestCase):
            @tornado.testing.gen_test
            def test_http_fetch(self):
                client = AsyncHTTPClient()
                response = yield client.fetch("http://www.tornadoweb.org")
                # Test contents of response
                self.assertIn("FriendFeed", response.body)

        # This test uses argument passing between self.stop and self.wait.
        class MyTestCase2(AsyncTestCase):
            def test_http_fetch(self):
                client = AsyncHTTPClient()
                client.fetch("http://www.tornadoweb.org/", self.stop)
                response = self.wait()
                # Test contents of response
                self.assertIn("FriendFeed", response.body)
    """

    def __init__(self, methodName: str = "runTest") -> None:
        super().__init__(methodName)
        self.__stopped = False
        self.__running = False
        self.__failure = None  # type: Optional[_ExcInfoTuple]
        self.__stop_args = None  # type: Any
        self.__timeout = None  # type: Optional[object]

        # It's easy to forget the @gen_test decorator, but if you do
        # the test will silently be ignored because nothing will consume
        # the generator.  Replace the test method with a wrapper that will
        # make sure it's not an undecorated generator.
        setattr(self, methodName, _TestMethodWrapper(getattr(self, methodName)))

        # Not used in this class itself, but used by @gen_test
        self._test_generator = None  # type: Optional[Union[Generator, Coroutine]]

    def setUp(self) -> None:
        setup_with_context_manager(self, warnings.catch_warnings())
        warnings.filterwarnings(
            "ignore",
            message="There is no current event loop",
            category=DeprecationWarning,
            module=r"tornado\..*",
        )
        super().setUp()
        # NOTE: this code attempts to navigate deprecation warnings introduced
        # in Python 3.10. The idea of an implicit current event loop is
        # deprecated in that version, with the intention that tests like this
        # explicitly create a new event loop and run on it. However, other
        # packages such as pytest-asyncio (as of version 0.16.0) still rely on
        # the implicit current event loop and we want to be compatible with them
        # (even when run on 3.10, but not, of course, on the future version of
        # python that removes the get/set_event_loop methods completely).
        #
        # Deprecation warnings were introduced inconsistently:
        # asyncio.get_event_loop warns, but
        # asyncio.get_event_loop_policy().get_event_loop does not. Similarly,
        # none of the set_event_loop methods warn, although comments on
        # https://bugs.python.org/issue39529 indicate that they are also
        # intended for future removal.
        #
        # Therefore, we first attempt to access the event loop with the
        # (non-warning) policy method, and if it fails, fall back to creating a
        # new event loop. We do not have effective test coverage of the
        # new event loop case; this will have to be watched when/if
        # get_event_loop is actually removed.
        self.should_close_asyncio_loop = False
        try:
            self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()
        except Exception:
            self.asyncio_loop = asyncio.new_event_loop()
            self.should_close_asyncio_loop = True

        async def get_loop() -> IOLoop:
            return self.get_new_ioloop()

        self.io_loop = self.asyncio_loop.run_until_complete(get_loop())
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)
            self.io_loop.make_current()

    def tearDown(self) -> None:
        # Native coroutines tend to produce warnings if they're not
        # allowed to run to completion. It's difficult to ensure that
        # this always happens in tests, so cancel any tasks that are
        # still pending by the time we get here.
        asyncio_loop = self.io_loop.asyncio_loop  # type: ignore
        if hasattr(asyncio, "all_tasks"):  # py37
            tasks = asyncio.all_tasks(asyncio_loop)  # type: ignore
        else:
            tasks = asyncio.Task.all_tasks(asyncio_loop)
        # Tasks that are done may still appear here and may contain
        # non-cancellation exceptions, so filter them out.
        tasks = [t for t in tasks if not t.done()]  # type: ignore
        for t in tasks:
            t.cancel()
        # Allow the tasks to run and finalize themselves (which means
        # raising a CancelledError inside the coroutine). This may
        # just transform the "task was destroyed but it is pending"
        # warning into a "uncaught CancelledError" warning, but
        # catching CancelledErrors in coroutines that may leak is
        # simpler than ensuring that no coroutines leak.
        if tasks:
            done, pending = self.io_loop.run_sync(lambda: asyncio.wait(tasks))
            assert not pending
            # If any task failed with anything but a CancelledError, raise it.
            for f in done:
                try:
                    f.result()
                except asyncio.CancelledError:
                    pass

        # Clean up Subprocess, so it can be used again with a new ioloop.
        Subprocess.uninitialize()
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)
            self.io_loop.clear_current()
        if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):
            # Try to clean up any file descriptors left open in the ioloop.
            # This avoids leaks, especially when tests are run repeatedly
            # in the same process with autoreload (because curl does not
            # set FD_CLOEXEC on its file descriptors)
            self.io_loop.close(all_fds=True)
        if self.should_close_asyncio_loop:
            self.asyncio_loop.close()
        super().tearDown()
        # In case an exception escaped or the StackContext caught an exception
        # when there wasn't a wait() to re-raise it, do so here.
        # This is our last chance to raise an exception in a way that the
        # unittest machinery understands.
        self.__rethrow()

    def get_new_ioloop(self) -> IOLoop:
        """Returns the `.IOLoop` to use for this test.

        By default, a new `.IOLoop` is created for each test.
        Subclasses may override this method to return
        `.IOLoop.current()` if it is not appropriate to use a new
        `.IOLoop` in each tests (for example, if there are global
        singletons using the default `.IOLoop`) or if a per-test event
        loop is being provided by another system (such as
        ``pytest-asyncio``).
        """
        return IOLoop(make_current=False)

    def _handle_exception(
        self, typ: Type[Exception], value: Exception, tb: TracebackType
    ) -> bool:
        if self.__failure is None:
            self.__failure = (typ, value, tb)
        else:
            app_log.error(
                "multiple unhandled exceptions in test", exc_info=(typ, value, tb)
            )
        self.stop()
        return True

    def __rethrow(self) -> None:
        if self.__failure is not None:
            failure = self.__failure
            self.__failure = None
            raise_exc_info(failure)

    def run(
        self, result: Optional[unittest.TestResult] = None
    ) -> Optional[unittest.TestResult]:
        ret = super().run(result)
        # As a last resort, if an exception escaped super.run() and wasn't
        # re-raised in tearDown, raise it here.  This will cause the
        # unittest run to fail messily, but that's better than silently
        # ignoring an error.
        self.__rethrow()
        return ret

    def stop(self, _arg: Any = None, **kwargs: Any) -> None:
        """Stops the `.IOLoop`, causing one pending (or future) call to `wait()`
        to return.

        Keyword arguments or a single positional argument passed to `stop()` are
        saved and will be returned by `wait()`.

        .. deprecated:: 5.1

           `stop` and `wait` are deprecated; use ``@gen_test`` instead.
        """
        assert _arg is None or not kwargs
        self.__stop_args = kwargs or _arg
        if self.__running:
            self.io_loop.stop()
            self.__running = False
        self.__stopped = True

    def wait(
        self,
        condition: Optional[Callable[..., bool]] = None,
        timeout: Optional[float] = None,
    ) -> Any:
        """Runs the `.IOLoop` until stop is called or timeout has passed.

        In the event of a timeout, an exception will be thrown. The
        default timeout is 5 seconds; it may be overridden with a
        ``timeout`` keyword argument or globally with the
        ``ASYNC_TEST_TIMEOUT`` environment variable.

        If ``condition`` is not ``None``, the `.IOLoop` will be restarted
        after `stop()` until ``condition()`` returns ``True``.

        .. versionchanged:: 3.1
           Added the ``ASYNC_TEST_TIMEOUT`` environment variable.

        .. deprecated:: 5.1

           `stop` and `wait` are deprecated; use ``@gen_test`` instead.
        """
        if timeout is None:
            timeout = get_async_test_timeout()

        if not self.__stopped:
            if timeout:

                def timeout_func() -> None:
                    try:
                        raise self.failureException(
                            "Async operation timed out after %s seconds" % timeout
                        )
                    except Exception:
                        self.__failure = sys.exc_info()
                    self.stop()

                self.__timeout = self.io_loop.add_timeout(
                    self.io_loop.time() + timeout, timeout_func
                )
            while True:
                self.__running = True
                self.io_loop.start()
                if self.__failure is not None or condition is None or condition():
                    break
            if self.__timeout is not None:
                self.io_loop.remove_timeout(self.__timeout)
                self.__timeout = None
        assert self.__stopped
        self.__stopped = False
        self.__rethrow()
        result = self.__stop_args
        self.__stop_args = None
        return result


class AsyncHTTPTestCase(AsyncTestCase):
    """A test case that starts up an HTTP server.

    Subclasses must override `get_app()`, which returns the
    `tornado.web.Application` (or other `.HTTPServer` callback) to be tested.
    Tests will typically use the provided ``self.http_client`` to fetch
    URLs from this server.

    Example, assuming the "Hello, world" example from the user guide is in
    ``hello.py``::

        import hello

        class TestHelloApp(AsyncHTTPTestCase):
            def get_app(self):
                return hello.make_app()

            def test_homepage(self):
                response = self.fetch('/')
                self.assertEqual(response.code, 200)
                self.assertEqual(response.body, 'Hello, world')

    That call to ``self.fetch()`` is equivalent to ::

        self.http_client.fetch(self.get_url('/'), self.stop)
        response = self.wait()

    which illustrates how AsyncTestCase can turn an asynchronous operation,
    like ``http_client.fetch()``, into a synchronous operation. If you need
    to do other asynchronous operations in tests, you'll probably need to use
    ``stop()`` and ``wait()`` yourself.
    """

    def setUp(self) -> None:
        super().setUp()
        sock, port = bind_unused_port()
        self.__port = port

        self.http_client = self.get_http_client()
        self._app = self.get_app()
        self.http_server = self.get_http_server()
        self.http_server.add_sockets([sock])

    def get_http_client(self) -> AsyncHTTPClient:
        return AsyncHTTPClient()

    def get_http_server(self) -> HTTPServer:
        return HTTPServer(self._app, **self.get_httpserver_options())

    def get_app(self) -> Application:
        """Should be overridden by subclasses to return a
        `tornado.web.Application` or other `.HTTPServer` callback.
        """
        raise NotImplementedError()

    def fetch(
        self, path: str, raise_error: bool = False, **kwargs: Any
    ) -> HTTPResponse:
        """Convenience method to synchronously fetch a URL.

        The given path will be appended to the local server's host and
        port.  Any additional keyword arguments will be passed directly to
        `.AsyncHTTPClient.fetch` (and so could be used to pass
        ``method="POST"``, ``body="..."``, etc).

        If the path begins with http:// or https://, it will be treated as a
        full URL and will be fetched as-is.

        If ``raise_error`` is ``True``, a `tornado.httpclient.HTTPError` will
        be raised if the response code is not 200. This is the same behavior
        as the ``raise_error`` argument to `.AsyncHTTPClient.fetch`, but
        the default is ``False`` here (it's ``True`` in `.AsyncHTTPClient`)
        because tests often need to deal with non-200 response codes.

        .. versionchanged:: 5.0
           Added support for absolute URLs.

        .. versionchanged:: 5.1

           Added the ``raise_error`` argument.

        .. deprecated:: 5.1

           This method currently turns any exception into an
           `.HTTPResponse` with status code 599. In Tornado 6.0,
           errors other than `tornado.httpclient.HTTPError` will be
           passed through, and ``raise_error=False`` will only
           suppress errors that would be raised due to non-200
           response codes.

        """
        if path.lower().startswith(("http://", "https://")):
            url = path
        else:
            url = self.get_url(path)
        return self.io_loop.run_sync(
            lambda: self.http_client.fetch(url, raise_error=raise_error, **kwargs),
            timeout=get_async_test_timeout(),
        )

    def get_httpserver_options(self) -> Dict[str, Any]:
        """May be overridden by subclasses to return additional
        keyword arguments for the server.
        """
        return {}

    def get_http_port(self) -> int:
        """Returns the port used by the server.

        A new port is chosen for each test.
        """
        return self.__port

    def get_protocol(self) -> str:
        return "http"

    def get_url(self, path: str) -> str:
        """Returns an absolute url for the given path on the test server."""
        return "%s://127.0.0.1:%s%s" % (self.get_protocol(), self.get_http_port(), path)

    def tearDown(self) -> None:
        self.http_server.stop()
        self.io_loop.run_sync(
            self.http_server.close_all_connections, timeout=get_async_test_timeout()
        )
        self.http_client.close()
        del self.http_server
        del self._app
        super().tearDown()


class AsyncHTTPSTestCase(AsyncHTTPTestCase):
    """A test case that starts an HTTPS server.

    Interface is generally the same as `AsyncHTTPTestCase`.
    """

    def get_http_client(self) -> AsyncHTTPClient:
        return AsyncHTTPClient(force_instance=True, defaults=dict(validate_cert=False))

    def get_httpserver_options(self) -> Dict[str, Any]:
        return dict(ssl_options=self.get_ssl_options())

    def get_ssl_options(self) -> Dict[str, Any]:
        """May be overridden by subclasses to select SSL options.

        By default includes a self-signed testing certificate.
        """
        return AsyncHTTPSTestCase.default_ssl_options()

    @staticmethod
    def default_ssl_options() -> Dict[str, Any]:
        # Testing keys were generated with:
        # openssl req -new -keyout tornado/test/test.key \
        #                     -out tornado/test/test.crt -nodes -days 3650 -x509
        module_dir = os.path.dirname(__file__)
        return dict(
            certfile=os.path.join(module_dir, "test", "test.crt"),
            keyfile=os.path.join(module_dir, "test", "test.key"),
        )

    def get_protocol(self) -> str:
        return "https"


@typing.overload
def gen_test(
    *, timeout: Optional[float] = None
) -> Callable[[Callable[..., Union[Generator, "Coroutine"]]], Callable[..., None]]:
    pass


@typing.overload  # noqa: F811
def gen_test(func: Callable[..., Union[Generator, "Coroutine"]]) -> Callable[..., None]:
    pass


def gen_test(  # noqa: F811
    func: Optional[Callable[..., Union[Generator, "Coroutine"]]] = None,
    timeout: Optional[float] = None,
) -> Union[
    Callable[..., None],
    Callable[[Callable[..., Union[Generator, "Coroutine"]]], Callable[..., None]],
]:
    """Testing equivalent of ``@gen.coroutine``, to be applied to test methods.

    ``@gen.coroutine`` cannot be used on tests because the `.IOLoop` is not
    already running.  ``@gen_test`` should be applied to test methods
    on subclasses of `AsyncTestCase`.

    Example::

        class MyTest(AsyncHTTPTestCase):
            @gen_test
            def test_something(self):
                response = yield self.http_client.fetch(self.get_url('/'))

    By default, ``@gen_test`` times out after 5 seconds. The timeout may be
    overridden globally with the ``ASYNC_TEST_TIMEOUT`` environment variable,
    or for each test with the ``timeout`` keyword argument::

        class MyTest(AsyncHTTPTestCase):
            @gen_test(timeout=10)
            def test_something_slow(self):
                response = yield self.http_client.fetch(self.get_url('/'))

    Note that ``@gen_test`` is incompatible with `AsyncTestCase.stop`,
    `AsyncTestCase.wait`, and `AsyncHTTPTestCase.fetch`. Use ``yield
    self.http_client.fetch(self.get_url())`` as shown above instead.

    .. versionadded:: 3.1
       The ``timeout`` argument and ``ASYNC_TEST_TIMEOUT`` environment
       variable.

    .. versionchanged:: 4.0
       The wrapper now passes along ``*args, **kwargs`` so it can be used
       on functions with arguments.

    """
    if timeout is None:
        timeout = get_async_test_timeout()

    def wrap(f: Callable[..., Union[Generator, "Coroutine"]]) -> Callable[..., None]:
        # Stack up several decorators to allow us to access the generator
        # object itself.  In the innermost wrapper, we capture the generator
        # and save it in an attribute of self.  Next, we run the wrapped
        # function through @gen.coroutine.  Finally, the coroutine is
        # wrapped again to make it synchronous with run_sync.
        #
        # This is a good case study arguing for either some sort of
        # extensibility in the gen decorators or cancellation support.
        @functools.wraps(f)
        def pre_coroutine(self, *args, **kwargs):
            # type: (AsyncTestCase, *Any, **Any) -> Union[Generator, Coroutine]
            # Type comments used to avoid pypy3 bug.
            result = f(self, *args, **kwargs)
            if isinstance(result, Generator) or inspect.iscoroutine(result):
                self._test_generator = result
            else:
                self._test_generator = None
            return result

        if inspect.iscoroutinefunction(f):
            coro = pre_coroutine
        else:
            coro = gen.coroutine(pre_coroutine)  # type: ignore[assignment]

        @functools.wraps(coro)
        def post_coroutine(self, *args, **kwargs):
            # type: (AsyncTestCase, *Any, **Any) -> None
            try:
                return self.io_loop.run_sync(
                    functools.partial(coro, self, *args, **kwargs), timeout=timeout
                )
            except TimeoutError as e:
                # run_sync raises an error with an unhelpful traceback.
                # If the underlying generator is still running, we can throw the
                # exception back into it so the stack trace is replaced by the
                # point where the test is stopped. The only reason the generator
                # would not be running would be if it were cancelled, which means
                # a native coroutine, so we can rely on the cr_running attribute.
                if self._test_generator is not None and getattr(
                    self._test_generator, "cr_running", True
                ):
                    self._test_generator.throw(e)
                    # In case the test contains an overly broad except
                    # clause, we may get back here.
                # Coroutine was stopped or didn't raise a useful stack trace,
                # so re-raise the original exception which is better than nothing.
                raise

        return post_coroutine

    if func is not None:
        # Used like:
        #     @gen_test
        #     def f(self):
        #         pass
        return wrap(func)
    else:
        # Used like @gen_test(timeout=10)
        return wrap


# Without this attribute, nosetests will try to run gen_test as a test
# anywhere it is imported.
gen_test.__test__ = False  # type: ignore


class ExpectLog(logging.Filter):
    """Context manager to capture and suppress expected log output.

    Useful to make tests of error conditions less noisy, while still
    leaving unexpected log entries visible.  *Not thread safe.*

    The attribute ``logged_stack`` is set to ``True`` if any exception
    stack trace was logged.

    Usage::

        with ExpectLog('tornado.application', "Uncaught exception"):
            error_response = self.fetch("/some_page")

    .. versionchanged:: 4.3
       Added the ``logged_stack`` attribute.
    """

    def __init__(
        self,
        logger: Union[logging.Logger, basestring_type],
        regex: str,
        required: bool = True,
        level: Optional[int] = None,
    ) -> None:
        """Constructs an ExpectLog context manager.

        :param logger: Logger object (or name of logger) to watch.  Pass an
            empty string to watch the root logger.
        :param regex: Regular expression to match.  Any log entries on the
            specified logger that match this regex will be suppressed.
        :param required: If true, an exception will be raised if the end of the
            ``with`` statement is reached without matching any log entries.
        :param level: A constant from the ``logging`` module indicating the
            expected log level. If this parameter is provided, only log messages
            at this level will be considered to match. Additionally, the
            supplied ``logger`` will have its level adjusted if necessary (for
            the duration of the ``ExpectLog`` to enable the expected message.

        .. versionchanged:: 6.1
           Added the ``level`` parameter.

        .. deprecated:: 6.3
           In Tornado 7.0, only ``WARNING`` and higher logging levels will be
           matched by default. To match ``INFO`` and lower levels, the ``level``
           argument must be used. This is changing to minimize differences
           between ``tornado.testing.main`` (which enables ``INFO`` logs by
           default) and most other test runners (including those in IDEs)
           which have ``INFO`` logs disabled by default.
        """
        if isinstance(logger, basestring_type):
            logger = logging.getLogger(logger)
        self.logger = logger
        self.regex = re.compile(regex)
        self.required = required
        # matched and deprecated_level_matched are a counter for the respective event.
        self.matched = 0
        self.deprecated_level_matched = 0
        self.logged_stack = False
        self.level = level
        self.orig_level = None  # type: Optional[int]

    def filter(self, record: logging.LogRecord) -> bool:
        if record.exc_info:
            self.logged_stack = True
        message = record.getMessage()
        if self.regex.match(message):
            if self.level is None and record.levelno < logging.WARNING:
                # We're inside the logging machinery here so generating a DeprecationWarning
                # here won't be reported cleanly (if warnings-as-errors is enabled, the error
                # just gets swallowed by the logging module), and even if it were it would
                # have the wrong stack trace. Just remember this fact and report it in
                # __exit__ instead.
                self.deprecated_level_matched += 1
            if self.level is not None and record.levelno != self.level:
                app_log.warning(
                    "Got expected log message %r at unexpected level (%s vs %s)"
                    % (message, logging.getLevelName(self.level), record.levelname)
                )
                return True
            self.matched += 1
            return False
        return True

    def __enter__(self) -> "ExpectLog":
        if self.level is not None and self.level < self.logger.getEffectiveLevel():
            self.orig_level = self.logger.level
            self.logger.setLevel(self.level)
        self.logger.addFilter(self)
        return self

    def __exit__(
        self,
        typ: "Optional[Type[BaseException]]",
        value: Optional[BaseException],
        tb: Optional[TracebackType],
    ) -> None:
        if self.orig_level is not None:
            self.logger.setLevel(self.orig_level)
        self.logger.removeFilter(self)
        if not typ and self.required and not self.matched:
            raise Exception("did not get expected log message")
        if (
            not typ
            and self.required
            and (self.deprecated_level_matched >= self.matched)
        ):
            warnings.warn(
                "ExpectLog matched at INFO or below without level argument",
                DeprecationWarning,
            )


# From https://nedbatchelder.com/blog/201508/using_context_managers_in_test_setup.html
def setup_with_context_manager(testcase: unittest.TestCase, cm: Any) -> Any:
    """Use a contextmanager to setUp a test case."""
    val = cm.__enter__()
    testcase.addCleanup(cm.__exit__, None, None, None)
    return val


def main(**kwargs: Any) -> None:
    """A simple test runner.

    This test runner is essentially equivalent to `unittest.main` from
    the standard library, but adds support for Tornado-style option
    parsing and log formatting. It is *not* necessary to use this
    `main` function to run tests using `AsyncTestCase`; these tests
    are self-contained and can run with any test runner.

    The easiest way to run a test is via the command line::

        python -m tornado.testing tornado.test.web_test

    See the standard library ``unittest`` module for ways in which
    tests can be specified.

    Projects with many tests may wish to define a test script like
    ``tornado/test/runtests.py``.  This script should define a method
    ``all()`` which returns a test suite and then call
    `tornado.testing.main()`.  Note that even when a test script is
    used, the ``all()`` test suite may be overridden by naming a
    single test on the command line::

        # Runs all tests
        python -m tornado.test.runtests
        # Runs one test
        python -m tornado.test.runtests tornado.test.web_test

    Additional keyword arguments passed through to ``unittest.main()``.
    For example, use ``tornado.testing.main(verbosity=2)``
    to show many test details as they are run.
    See http://docs.python.org/library/unittest.html#unittest.main
    for full argument list.

    .. versionchanged:: 5.0

       This function produces no output of its own; only that produced
       by the `unittest` module (previously it would add a PASS or FAIL
       log message).
    """
    from tornado.options import define, options, parse_command_line

    define(
        "exception_on_interrupt",
        type=bool,
        default=True,
        help=(
            "If true (default), ctrl-c raises a KeyboardInterrupt "
            "exception.  This prints a stack trace but cannot interrupt "
            "certain operations.  If false, the process is more reliably "
            "killed, but does not print a stack trace."
        ),
    )

    # support the same options as unittest's command-line interface
    define("verbose", type=bool)
    define("quiet", type=bool)
    define("failfast", type=bool)
    define("catch", type=bool)
    define("buffer", type=bool)

    argv = [sys.argv[0]] + parse_command_line(sys.argv)

    if not options.exception_on_interrupt:
        signal.signal(signal.SIGINT, signal.SIG_DFL)

    if options.verbose is not None:
        kwargs["verbosity"] = 2
    if options.quiet is not None:
        kwargs["verbosity"] = 0
    if options.failfast is not None:
        kwargs["failfast"] = True
    if options.catch is not None:
        kwargs["catchbreak"] = True
    if options.buffer is not None:
        kwargs["buffer"] = True

    if __name__ == "__main__" and len(argv) == 1:
        print("No tests specified", file=sys.stderr)
        sys.exit(1)
    # In order to be able to run tests by their fully-qualified name
    # on the command line without importing all tests here,
    # module must be set to None.  Python 3.2's unittest.main ignores
    # defaultTest if no module is given (it tries to do its own
    # test discovery, which is incompatible with auto2to3), so don't
    # set module if we're not asking for a specific test.
    if len(argv) > 1:
        unittest.main(module=None, argv=argv, **kwargs)  # type: ignore
    else:
        unittest.main(defaultTest="all", argv=argv, **kwargs)


if __name__ == "__main__":
    main()

```
--- END FILE ---

Please first localize the bug based on the issue statement, and then generate *SEARCH/REPLACE* edits to fix the issue.

Every *SEARCH/REPLACE* edit must use this format:
1. The file path
2. The start of search block: <<<<<<< SEARCH
3. A contiguous chunk of lines to search for in the existing source code
4. The dividing line: =======
5. The lines to replace into the source code
6. The end of the replace block: >>>>>>> REPLACE

Here is an example:

```python
### mathweb/flask/app.py
<<<<<<< SEARCH
from flask import Flask
=======
import math
from flask import Flask
>>>>>>> REPLACE
```
</solution>

Please note that the *SEARCH/REPLACE* edit REQUIRES PROPER INDENTATION. If you would like to add the line '        print(x)', you must fully write that out, with all those spaces before the code!
Wrap each *SEARCH/REPLACE* edit in a code block as shown in the example above. If you have multiple *SEARCH/REPLACE* edits, use a separate code block for each one.
user