{
    "file_diffs": [
        {
            "old_file_content": "#!/usr/bin/env python\n\nfrom __future__ import absolute_import, division, print_function, with_statement\n\nfrom tornado import gen\nfrom tornado.testing import AsyncTestCase, gen_test\nfrom tornado.test.util import unittest\n\n\nclass AsyncTestCaseTest(AsyncTestCase):\n    def test_exception_in_callback(self):\n        self.io_loop.add_callback(lambda: 1 / 0)\n        try:\n            self.wait()\n            self.fail(\"did not get expected exception\")\n        except ZeroDivisionError:\n            pass\n\n    def test_subsequent_wait_calls(self):\n        \"\"\"\n        This test makes sure that a second call to wait()\n        clears the first timeout.\n        \"\"\"\n        self.io_loop.add_timeout(self.io_loop.time() + 0.01, self.stop)\n        self.wait(timeout=0.02)\n        self.io_loop.add_timeout(self.io_loop.time() + 0.03, self.stop)\n        self.wait(timeout=0.15)\n\n\nclass SetUpTearDownTest(unittest.TestCase):\n    def test_set_up_tear_down(self):\n        \"\"\"\n        This test makes sure that AsyncTestCase calls super methods for\n        setUp and tearDown.\n\n        InheritBoth is a subclass of both AsyncTestCase and\n        SetUpTearDown, with the ordering so that the super of\n        AsyncTestCase will be SetUpTearDown.\n        \"\"\"\n        events = []\n        result = unittest.TestResult()\n\n        class SetUpTearDown(unittest.TestCase):\n            def setUp(self):\n                events.append('setUp')\n\n            def tearDown(self):\n                events.append('tearDown')\n\n        class InheritBoth(AsyncTestCase, SetUpTearDown):\n            def test(self):\n                events.append('test')\n\n        InheritBoth('test').run(result)\n        expected = ['setUp', 'test', 'tearDown']\n        self.assertEqual(expected, events)\n\n\nclass GenTest(AsyncTestCase):\n    def setUp(self):\n        super(GenTest, self).setUp()\n        self.finished = False\n\n    def tearDown(self):\n        self.assertTrue(self.finished)\n        super(GenTest, self).tearDown()\n\n    @gen_test\n    def test_sync(self):\n        self.finished = True\n\n    @gen_test\n    def test_async(self):\n        yield gen.Task(self.io_loop.add_callback)\n        self.finished = True\n\nif __name__ == '__main__':\n    unittest.main()\n",
            "new_file_content": "#!/usr/bin/env python\n\nfrom __future__ import absolute_import, division, print_function, with_statement\n\nfrom tornado import gen, ioloop\nfrom tornado.testing import AsyncTestCase, gen_test\nfrom tornado.test.util import unittest\n\nimport functools\nimport os\n\n\nclass AsyncTestCaseTest(AsyncTestCase):\n    def test_exception_in_callback(self):\n        self.io_loop.add_callback(lambda: 1 / 0)\n        try:\n            self.wait()\n            self.fail(\"did not get expected exception\")\n        except ZeroDivisionError:\n            pass\n\n    def test_subsequent_wait_calls(self):\n        \"\"\"\n        This test makes sure that a second call to wait()\n        clears the first timeout.\n        \"\"\"\n        self.io_loop.add_timeout(self.io_loop.time() + 0.01, self.stop)\n        self.wait(timeout=0.02)\n        self.io_loop.add_timeout(self.io_loop.time() + 0.03, self.stop)\n        self.wait(timeout=0.15)\n\n\nclass SetUpTearDownTest(unittest.TestCase):\n    def test_set_up_tear_down(self):\n        \"\"\"\n        This test makes sure that AsyncTestCase calls super methods for\n        setUp and tearDown.\n\n        InheritBoth is a subclass of both AsyncTestCase and\n        SetUpTearDown, with the ordering so that the super of\n        AsyncTestCase will be SetUpTearDown.\n        \"\"\"\n        events = []\n        result = unittest.TestResult()\n\n        class SetUpTearDown(unittest.TestCase):\n            def setUp(self):\n                events.append('setUp')\n\n            def tearDown(self):\n                events.append('tearDown')\n\n        class InheritBoth(AsyncTestCase, SetUpTearDown):\n            def test(self):\n                events.append('test')\n\n        InheritBoth('test').run(result)\n        expected = ['setUp', 'test', 'tearDown']\n        self.assertEqual(expected, events)\n\n\nclass GenTest(AsyncTestCase):\n    def setUp(self):\n        super(GenTest, self).setUp()\n        self.finished = False\n\n    def tearDown(self):\n        self.assertTrue(self.finished)\n        super(GenTest, self).tearDown()\n\n    @gen_test\n    def test_sync(self):\n        self.finished = True\n\n    @gen_test\n    def test_async(self):\n        yield gen.Task(self.io_loop.add_callback)\n        self.finished = True\n\n    def test_timeout(self):\n        # Set a short timeout and exceed it.\n        @gen_test(timeout=0.1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)\n\n        with self.assertRaises(ioloop.TimeoutError):\n            test(self)\n\n        self.finished = True\n\n    def test_no_timeout(self):\n        # A test that does not exceed its timeout should succeed.\n        @gen_test(timeout=1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 0.1)\n\n        test(self)\n        self.finished = True\n\n    def test_timeout_environment_variable(self):\n        time = self.io_loop.time\n        add_timeout = self.io_loop.add_timeout\n        old_timeout = os.environ.get('TIMEOUT')\n        try:\n            os.environ['TIMEOUT'] = '0.1'\n\n            @gen_test(timeout=0.5)\n            def test_long_timeout(self):\n                yield gen.Task(add_timeout, time() + 0.25)\n\n            # Uses provided timeout of 0.5 seconds, doesn't time out.\n            self.io_loop.run_sync(\n                functools.partial(test_long_timeout, self))\n\n            @gen_test(timeout=0.01)\n            def test_short_timeout(self):\n                yield gen.Task(add_timeout, time() + 1)\n\n            # Uses environment TIMEOUT of 0.1, times out.\n            with self.assertRaises(ioloop.TimeoutError):\n                test_short_timeout(self)\n\n            self.finished = True\n        finally:\n            if old_timeout is None:\n                del os.environ['TIMEOUT']\n            else:\n                os.environ['TIMEOUT'] = old_timeout\n\nif __name__ == '__main__':\n    unittest.main()\n",
            "header": {
                "file": {
                    "path": "tornado/test/testing_test.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "dc9e0866",
                "new_commit_hash": "40c2fe53",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/tornado/test/testing_test.py"
            },
            "plus_file": {
                "path": "b/tornado/test/testing_test.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 2,
                            "length": 10
                        },
                        "new_range": {
                            "start": 2,
                            "length": 13
                        },
                        "section": ""
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "from __future__ import absolute_import, division, print_function, with_statement",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "from tornado import gen",
                                "type": "deleted"
                            },
                            {
                                "content": "from tornado import gen, ioloop",
                                "type": "added"
                            },
                            {
                                "content": "from tornado.testing import AsyncTestCase, gen_test",
                                "type": "context"
                            },
                            {
                                "content": "from tornado.test.util import unittest",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "import functools",
                                "type": "added"
                            },
                            {
                                "content": "import os",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "class AsyncTestCaseTest(AsyncTestCase):",
                                "type": "context"
                            },
                            {
                                "content": "    def test_exception_in_callback(self):",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [],
                    "added_entities": [
                        {
                            "file_name": "tornado/test/testing_test.py",
                            "type": "import",
                            "name": "functools",
                            "content": "import functools",
                            "ast_type_str": "Import",
                            "start_lineno": 9,
                            "end_lineno": 9,
                            "parent": null
                        },
                        {
                            "file_name": "tornado/test/testing_test.py",
                            "type": "statement",
                            "name": "",
                            "content": "from tornado import gen, ioloop",
                            "ast_type_str": "ImportFrom",
                            "start_lineno": 5,
                            "end_lineno": 5,
                            "parent": null
                        },
                        {
                            "file_name": "tornado/test/testing_test.py",
                            "type": "import",
                            "name": "os",
                            "content": "import os",
                            "ast_type_str": "Import",
                            "start_lineno": 10,
                            "end_lineno": 10,
                            "parent": null
                        }
                    ],
                    "deleted_entities": [
                        {
                            "file_name": "tornado/test/testing_test.py",
                            "type": "statement",
                            "name": "",
                            "content": "from tornado import gen",
                            "ast_type_str": "ImportFrom",
                            "start_lineno": 5,
                            "end_lineno": 5,
                            "parent": null
                        }
                    ]
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 74,
                            "length": 5
                        },
                        "new_range": {
                            "start": 77,
                            "length": 55
                        },
                        "section": "class GenTest(AsyncTestCase):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        yield gen.Task(self.io_loop.add_callback)",
                                "type": "context"
                            },
                            {
                                "content": "        self.finished = True",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    def test_timeout(self):",
                                "type": "added"
                            },
                            {
                                "content": "        # Set a short timeout and exceed it.",
                                "type": "added"
                            },
                            {
                                "content": "        @gen_test(timeout=0.1)",
                                "type": "added"
                            },
                            {
                                "content": "        def test(self):",
                                "type": "added"
                            },
                            {
                                "content": "            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "        with self.assertRaises(ioloop.TimeoutError):",
                                "type": "added"
                            },
                            {
                                "content": "            test(self)",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "        self.finished = True",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "    def test_no_timeout(self):",
                                "type": "added"
                            },
                            {
                                "content": "        # A test that does not exceed its timeout should succeed.",
                                "type": "added"
                            },
                            {
                                "content": "        @gen_test(timeout=1)",
                                "type": "added"
                            },
                            {
                                "content": "        def test(self):",
                                "type": "added"
                            },
                            {
                                "content": "            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 0.1)",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "        test(self)",
                                "type": "added"
                            },
                            {
                                "content": "        self.finished = True",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "    def test_timeout_environment_variable(self):",
                                "type": "added"
                            },
                            {
                                "content": "        time = self.io_loop.time",
                                "type": "added"
                            },
                            {
                                "content": "        add_timeout = self.io_loop.add_timeout",
                                "type": "added"
                            },
                            {
                                "content": "        old_timeout = os.environ.get('TIMEOUT')",
                                "type": "added"
                            },
                            {
                                "content": "        try:",
                                "type": "added"
                            },
                            {
                                "content": "            os.environ['TIMEOUT'] = '0.1'",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "            @gen_test(timeout=0.5)",
                                "type": "added"
                            },
                            {
                                "content": "            def test_long_timeout(self):",
                                "type": "added"
                            },
                            {
                                "content": "                yield gen.Task(add_timeout, time() + 0.25)",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "            # Uses provided timeout of 0.5 seconds, doesn't time out.",
                                "type": "added"
                            },
                            {
                                "content": "            self.io_loop.run_sync(",
                                "type": "added"
                            },
                            {
                                "content": "                functools.partial(test_long_timeout, self))",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "            @gen_test(timeout=0.01)",
                                "type": "added"
                            },
                            {
                                "content": "            def test_short_timeout(self):",
                                "type": "added"
                            },
                            {
                                "content": "                yield gen.Task(add_timeout, time() + 1)",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "            # Uses environment TIMEOUT of 0.1, times out.",
                                "type": "added"
                            },
                            {
                                "content": "            with self.assertRaises(ioloop.TimeoutError):",
                                "type": "added"
                            },
                            {
                                "content": "                test_short_timeout(self)",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "            self.finished = True",
                                "type": "added"
                            },
                            {
                                "content": "        finally:",
                                "type": "added"
                            },
                            {
                                "content": "            if old_timeout is None:",
                                "type": "added"
                            },
                            {
                                "content": "                del os.environ['TIMEOUT']",
                                "type": "added"
                            },
                            {
                                "content": "            else:",
                                "type": "added"
                            },
                            {
                                "content": "                os.environ['TIMEOUT'] = old_timeout",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "if __name__ == '__main__':",
                                "type": "context"
                            },
                            {
                                "content": "    unittest.main()",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/test/testing_test.py",
                            "type": "class",
                            "name": "GenTest",
                            "content": "class GenTest(AsyncTestCase):\n    def setUp(self):\n        super(GenTest, self).setUp()\n        self.finished = False\n\n    def tearDown(self):\n        self.assertTrue(self.finished)\n        super(GenTest, self).tearDown()\n\n    @gen_test\n    def test_sync(self):\n        self.finished = True\n\n    @gen_test\n    def test_async(self):\n        yield gen.Task(self.io_loop.add_callback)\n        self.finished = True\n\n    def test_timeout(self):\n        # Set a short timeout and exceed it.\n        @gen_test(timeout=0.1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)\n\n        with self.assertRaises(ioloop.TimeoutError):\n            test(self)\n\n        self.finished = True\n\n    def test_no_timeout(self):\n        # A test that does not exceed its timeout should succeed.\n        @gen_test(timeout=1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 0.1)\n\n        test(self)\n        self.finished = True\n\n    def test_timeout_environment_variable(self):\n        time = self.io_loop.time\n        add_timeout = self.io_loop.add_timeout\n        old_timeout = os.environ.get('TIMEOUT')\n        try:\n            os.environ['TIMEOUT'] = '0.1'\n\n            @gen_test(timeout=0.5)\n            def test_long_timeout(self):\n                yield gen.Task(add_timeout, time() + 0.25)\n\n            # Uses provided timeout of 0.5 seconds, doesn't time out.\n            self.io_loop.run_sync(\n                functools.partial(test_long_timeout, self))\n\n            @gen_test(timeout=0.01)\n            def test_short_timeout(self):\n                yield gen.Task(add_timeout, time() + 1)\n\n            # Uses environment TIMEOUT of 0.1, times out.\n            with self.assertRaises(ioloop.TimeoutError):\n                test_short_timeout(self)\n\n            self.finished = True\n        finally:\n            if old_timeout is None:\n                del os.environ['TIMEOUT']\n            else:\n                os.environ['TIMEOUT'] = old_timeout",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 62,
                            "end_lineno": 128,
                            "parent": null
                        }
                    ],
                    "added_entities": [
                        {
                            "file_name": "tornado/test/testing_test.py",
                            "type": "method",
                            "name": "GenTest.test_timeout_environment_variable",
                            "content": "class GenTest(AsyncTestCase):\n    def setUp(self):\n        super(GenTest, self).setUp()\n        self.finished = False\n\n    def tearDown(self):\n        self.assertTrue(self.finished)\n        super(GenTest, self).tearDown()\n\n    @gen_test\n    def test_sync(self):\n        self.finished = True\n\n    @gen_test\n    def test_async(self):\n        yield gen.Task(self.io_loop.add_callback)\n        self.finished = True\n\n    def test_timeout(self):\n        # Set a short timeout and exceed it.\n        @gen_test(timeout=0.1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)\n\n        with self.assertRaises(ioloop.TimeoutError):\n            test(self)\n\n        self.finished = True\n\n    def test_no_timeout(self):\n        # A test that does not exceed its timeout should succeed.\n        @gen_test(timeout=1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 0.1)\n\n        test(self)\n        self.finished = True\n\n    def test_timeout_environment_variable(self):\n        time = self.io_loop.time\n        add_timeout = self.io_loop.add_timeout\n        old_timeout = os.environ.get('TIMEOUT')\n        try:\n            os.environ['TIMEOUT'] = '0.1'\n\n            @gen_test(timeout=0.5)\n            def test_long_timeout(self):\n                yield gen.Task(add_timeout, time() + 0.25)\n\n            # Uses provided timeout of 0.5 seconds, doesn't time out.\n            self.io_loop.run_sync(\n                functools.partial(test_long_timeout, self))\n\n            @gen_test(timeout=0.01)\n            def test_short_timeout(self):\n                yield gen.Task(add_timeout, time() + 1)\n\n            # Uses environment TIMEOUT of 0.1, times out.\n            with self.assertRaises(ioloop.TimeoutError):\n                test_short_timeout(self)\n\n            self.finished = True\n        finally:\n            if old_timeout is None:\n                del os.environ['TIMEOUT']\n            else:\n                os.environ['TIMEOUT'] = old_timeout",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 100,
                            "end_lineno": 128,
                            "parent": {
                                "file_name": "tornado/test/testing_test.py",
                                "type": "class",
                                "name": "GenTest",
                                "content": "class GenTest(AsyncTestCase):\n    def setUp(self):\n        super(GenTest, self).setUp()\n        self.finished = False\n\n    def tearDown(self):\n        self.assertTrue(self.finished)\n        super(GenTest, self).tearDown()\n\n    @gen_test\n    def test_sync(self):\n        self.finished = True\n\n    @gen_test\n    def test_async(self):\n        yield gen.Task(self.io_loop.add_callback)\n        self.finished = True\n\n    def test_timeout(self):\n        # Set a short timeout and exceed it.\n        @gen_test(timeout=0.1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)\n\n        with self.assertRaises(ioloop.TimeoutError):\n            test(self)\n\n        self.finished = True\n\n    def test_no_timeout(self):\n        # A test that does not exceed its timeout should succeed.\n        @gen_test(timeout=1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 0.1)\n\n        test(self)\n        self.finished = True\n\n    def test_timeout_environment_variable(self):\n        time = self.io_loop.time\n        add_timeout = self.io_loop.add_timeout\n        old_timeout = os.environ.get('TIMEOUT')\n        try:\n            os.environ['TIMEOUT'] = '0.1'\n\n            @gen_test(timeout=0.5)\n            def test_long_timeout(self):\n                yield gen.Task(add_timeout, time() + 0.25)\n\n            # Uses provided timeout of 0.5 seconds, doesn't time out.\n            self.io_loop.run_sync(\n                functools.partial(test_long_timeout, self))\n\n            @gen_test(timeout=0.01)\n            def test_short_timeout(self):\n                yield gen.Task(add_timeout, time() + 1)\n\n            # Uses environment TIMEOUT of 0.1, times out.\n            with self.assertRaises(ioloop.TimeoutError):\n                test_short_timeout(self)\n\n            self.finished = True\n        finally:\n            if old_timeout is None:\n                del os.environ['TIMEOUT']\n            else:\n                os.environ['TIMEOUT'] = old_timeout",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 62,
                                "end_lineno": 128,
                                "parent": null
                            }
                        },
                        {
                            "file_name": "tornado/test/testing_test.py",
                            "type": "method",
                            "name": "GenTest.test_timeout",
                            "content": "class GenTest(AsyncTestCase):\n    def setUp(self):\n        super(GenTest, self).setUp()\n        self.finished = False\n\n    def tearDown(self):\n        self.assertTrue(self.finished)\n        super(GenTest, self).tearDown()\n\n    @gen_test\n    def test_sync(self):\n        self.finished = True\n\n    @gen_test\n    def test_async(self):\n        yield gen.Task(self.io_loop.add_callback)\n        self.finished = True\n\n    def test_timeout(self):\n        # Set a short timeout and exceed it.\n        @gen_test(timeout=0.1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)\n\n        with self.assertRaises(ioloop.TimeoutError):\n            test(self)\n\n        self.finished = True\n\n    def test_no_timeout(self):\n        # A test that does not exceed its timeout should succeed.\n        @gen_test(timeout=1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 0.1)\n\n        test(self)\n        self.finished = True\n\n    def test_timeout_environment_variable(self):\n        time = self.io_loop.time\n        add_timeout = self.io_loop.add_timeout\n        old_timeout = os.environ.get('TIMEOUT')\n        try:\n            os.environ['TIMEOUT'] = '0.1'\n\n            @gen_test(timeout=0.5)\n            def test_long_timeout(self):\n                yield gen.Task(add_timeout, time() + 0.25)\n\n            # Uses provided timeout of 0.5 seconds, doesn't time out.\n            self.io_loop.run_sync(\n                functools.partial(test_long_timeout, self))\n\n            @gen_test(timeout=0.01)\n            def test_short_timeout(self):\n                yield gen.Task(add_timeout, time() + 1)\n\n            # Uses environment TIMEOUT of 0.1, times out.\n            with self.assertRaises(ioloop.TimeoutError):\n                test_short_timeout(self)\n\n            self.finished = True\n        finally:\n            if old_timeout is None:\n                del os.environ['TIMEOUT']\n            else:\n                os.environ['TIMEOUT'] = old_timeout",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 80,
                            "end_lineno": 89,
                            "parent": {
                                "file_name": "tornado/test/testing_test.py",
                                "type": "class",
                                "name": "GenTest",
                                "content": "class GenTest(AsyncTestCase):\n    def setUp(self):\n        super(GenTest, self).setUp()\n        self.finished = False\n\n    def tearDown(self):\n        self.assertTrue(self.finished)\n        super(GenTest, self).tearDown()\n\n    @gen_test\n    def test_sync(self):\n        self.finished = True\n\n    @gen_test\n    def test_async(self):\n        yield gen.Task(self.io_loop.add_callback)\n        self.finished = True\n\n    def test_timeout(self):\n        # Set a short timeout and exceed it.\n        @gen_test(timeout=0.1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)\n\n        with self.assertRaises(ioloop.TimeoutError):\n            test(self)\n\n        self.finished = True\n\n    def test_no_timeout(self):\n        # A test that does not exceed its timeout should succeed.\n        @gen_test(timeout=1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 0.1)\n\n        test(self)\n        self.finished = True\n\n    def test_timeout_environment_variable(self):\n        time = self.io_loop.time\n        add_timeout = self.io_loop.add_timeout\n        old_timeout = os.environ.get('TIMEOUT')\n        try:\n            os.environ['TIMEOUT'] = '0.1'\n\n            @gen_test(timeout=0.5)\n            def test_long_timeout(self):\n                yield gen.Task(add_timeout, time() + 0.25)\n\n            # Uses provided timeout of 0.5 seconds, doesn't time out.\n            self.io_loop.run_sync(\n                functools.partial(test_long_timeout, self))\n\n            @gen_test(timeout=0.01)\n            def test_short_timeout(self):\n                yield gen.Task(add_timeout, time() + 1)\n\n            # Uses environment TIMEOUT of 0.1, times out.\n            with self.assertRaises(ioloop.TimeoutError):\n                test_short_timeout(self)\n\n            self.finished = True\n        finally:\n            if old_timeout is None:\n                del os.environ['TIMEOUT']\n            else:\n                os.environ['TIMEOUT'] = old_timeout",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 62,
                                "end_lineno": 128,
                                "parent": null
                            }
                        },
                        {
                            "file_name": "tornado/test/testing_test.py",
                            "type": "method",
                            "name": "GenTest.test_no_timeout",
                            "content": "class GenTest(AsyncTestCase):\n    def setUp(self):\n        super(GenTest, self).setUp()\n        self.finished = False\n\n    def tearDown(self):\n        self.assertTrue(self.finished)\n        super(GenTest, self).tearDown()\n\n    @gen_test\n    def test_sync(self):\n        self.finished = True\n\n    @gen_test\n    def test_async(self):\n        yield gen.Task(self.io_loop.add_callback)\n        self.finished = True\n\n    def test_timeout(self):\n        # Set a short timeout and exceed it.\n        @gen_test(timeout=0.1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)\n\n        with self.assertRaises(ioloop.TimeoutError):\n            test(self)\n\n        self.finished = True\n\n    def test_no_timeout(self):\n        # A test that does not exceed its timeout should succeed.\n        @gen_test(timeout=1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 0.1)\n\n        test(self)\n        self.finished = True\n\n    def test_timeout_environment_variable(self):\n        time = self.io_loop.time\n        add_timeout = self.io_loop.add_timeout\n        old_timeout = os.environ.get('TIMEOUT')\n        try:\n            os.environ['TIMEOUT'] = '0.1'\n\n            @gen_test(timeout=0.5)\n            def test_long_timeout(self):\n                yield gen.Task(add_timeout, time() + 0.25)\n\n            # Uses provided timeout of 0.5 seconds, doesn't time out.\n            self.io_loop.run_sync(\n                functools.partial(test_long_timeout, self))\n\n            @gen_test(timeout=0.01)\n            def test_short_timeout(self):\n                yield gen.Task(add_timeout, time() + 1)\n\n            # Uses environment TIMEOUT of 0.1, times out.\n            with self.assertRaises(ioloop.TimeoutError):\n                test_short_timeout(self)\n\n            self.finished = True\n        finally:\n            if old_timeout is None:\n                del os.environ['TIMEOUT']\n            else:\n                os.environ['TIMEOUT'] = old_timeout",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 91,
                            "end_lineno": 98,
                            "parent": {
                                "file_name": "tornado/test/testing_test.py",
                                "type": "class",
                                "name": "GenTest",
                                "content": "class GenTest(AsyncTestCase):\n    def setUp(self):\n        super(GenTest, self).setUp()\n        self.finished = False\n\n    def tearDown(self):\n        self.assertTrue(self.finished)\n        super(GenTest, self).tearDown()\n\n    @gen_test\n    def test_sync(self):\n        self.finished = True\n\n    @gen_test\n    def test_async(self):\n        yield gen.Task(self.io_loop.add_callback)\n        self.finished = True\n\n    def test_timeout(self):\n        # Set a short timeout and exceed it.\n        @gen_test(timeout=0.1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)\n\n        with self.assertRaises(ioloop.TimeoutError):\n            test(self)\n\n        self.finished = True\n\n    def test_no_timeout(self):\n        # A test that does not exceed its timeout should succeed.\n        @gen_test(timeout=1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 0.1)\n\n        test(self)\n        self.finished = True\n\n    def test_timeout_environment_variable(self):\n        time = self.io_loop.time\n        add_timeout = self.io_loop.add_timeout\n        old_timeout = os.environ.get('TIMEOUT')\n        try:\n            os.environ['TIMEOUT'] = '0.1'\n\n            @gen_test(timeout=0.5)\n            def test_long_timeout(self):\n                yield gen.Task(add_timeout, time() + 0.25)\n\n            # Uses provided timeout of 0.5 seconds, doesn't time out.\n            self.io_loop.run_sync(\n                functools.partial(test_long_timeout, self))\n\n            @gen_test(timeout=0.01)\n            def test_short_timeout(self):\n                yield gen.Task(add_timeout, time() + 1)\n\n            # Uses environment TIMEOUT of 0.1, times out.\n            with self.assertRaises(ioloop.TimeoutError):\n                test_short_timeout(self)\n\n            self.finished = True\n        finally:\n            if old_timeout is None:\n                del os.environ['TIMEOUT']\n            else:\n                os.environ['TIMEOUT'] = old_timeout",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 62,
                                "end_lineno": 128,
                                "parent": null
                            }
                        }
                    ],
                    "deleted_entities": []
                }
            ]
        },
        {
            "old_file_content": "#!/usr/bin/env python\n\"\"\"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` and `LogTrapTestCase`: 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\nfrom __future__ import absolute_import, division, print_function, with_statement\n\ntry:\n    from tornado import gen\n    from tornado.httpclient import AsyncHTTPClient\n    from tornado.httpserver import HTTPServer\n    from tornado.simple_httpclient import SimpleAsyncHTTPClient\n    from tornado.ioloop import IOLoop\n    from tornado import netutil\nexcept ImportError:\n    # These modules are not importable on app engine.  Parts of this module\n    # won't work, but e.g. LogTrapTestCase and main() will.\n    AsyncHTTPClient = None\n    gen = None\n    HTTPServer = None\n    IOLoop = None\n    netutil = None\n    SimpleAsyncHTTPClient = None\nfrom tornado.log import gen_log\nfrom tornado.stack_context import ExceptionStackContext\nfrom tornado.util import raise_exc_info, basestring_type\nimport functools\nimport logging\nimport os\nimport re\nimport signal\nimport socket\nimport sys\n\ntry:\n    from cStringIO import StringIO  # py2\nexcept ImportError:\n    from io import StringIO  # py3\n\n# Tornado's own test suite requires the updated unittest module\n# (either py27+ or unittest2) so tornado.test.util enforces\n# this requirement, but for other users of tornado.testing we want\n# to allow the older version if unitest2 is not available.\ntry:\n    import unittest2 as unittest\nexcept ImportError:\n    import unittest\n\n_next_port = 10000\n\n\ndef get_unused_port():\n    \"\"\"Returns a (hopefully) unused port number.\n\n    This function does not guarantee that the port it returns is available,\n    only that a series of get_unused_port calls in a single process return\n    distinct ports.\n\n    **Deprecated**.  Use bind_unused_port instead, which is guaranteed\n    to find an unused port.\n    \"\"\"\n    global _next_port\n    port = _next_port\n    _next_port = _next_port + 1\n    return port\n\n\ndef bind_unused_port():\n    \"\"\"Binds a server socket to an available port on localhost.\n\n    Returns a tuple (socket, port).\n    \"\"\"\n    [sock] = netutil.bind_sockets(0, 'localhost', family=socket.AF_INET)\n    port = sock.getsockname()[1]\n    return sock, port\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 class provides\n    the `stop()` and `wait()` methods for this purpose.  The test\n    method itself must call ``self.wait()``, and asynchronous\n    callbacks should call ``self.stop()`` to signal completion.\n    Alternately, the `gen_test` decorator can be used to use yield points\n    from the `tornado.gen` module.\n\n    By default, a new `.IOLoop` is constructed for each test and is available\n    as ``self.io_loop``.  This `.IOLoop` should be used in the construction of\n    HTTP clients/servers, etc.  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 argument passing between self.stop and self.wait.\n        class MyTestCase(AsyncTestCase):\n            def test_http_fetch(self):\n                client = AsyncHTTPClient(self.io_loop)\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        # This test uses an explicit callback-based style.\n        class MyTestCase2(AsyncTestCase):\n            def test_http_fetch(self):\n                client = AsyncHTTPClient(self.io_loop)\n                client.fetch(\"http://www.tornadoweb.org/\", self.handle_fetch)\n                self.wait()\n\n            def handle_fetch(self, response):\n                # Test contents of response (failures and exceptions here\n                # will cause self.wait() to throw an exception and end the\n                # test).\n                # Exceptions thrown here are magically propagated to\n                # self.wait() in test_http_fetch() via stack_context.\n                self.assertIn(\"FriendFeed\", response.body)\n                self.stop()\n    \"\"\"\n    def __init__(self, *args, **kwargs):\n        super(AsyncTestCase, self).__init__(*args, **kwargs)\n        self.__stopped = False\n        self.__running = False\n        self.__failure = None\n        self.__stop_args = None\n        self.__timeout = None\n\n    def setUp(self):\n        super(AsyncTestCase, self).setUp()\n        self.io_loop = self.get_new_ioloop()\n        self.io_loop.make_current()\n\n    def tearDown(self):\n        self.io_loop.clear_current()\n        if (not IOLoop.initialized() or\n                self.io_loop is not IOLoop.instance()):\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(AsyncTestCase, self).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):\n        \"\"\"Creates a new `.IOLoop` for this test.  May be overridden in\n        subclasses for tests that require a specific `.IOLoop` (usually\n        the singleton `.IOLoop.instance()`).\n        \"\"\"\n        return IOLoop()\n\n    def _handle_exception(self, typ, value, tb):\n        self.__failure = sys.exc_info()\n        self.stop()\n        return True\n\n    def __rethrow(self):\n        if self.__failure is not None:\n            failure = self.__failure\n            self.__failure = None\n            raise_exc_info(failure)\n\n    def run(self, result=None):\n        with ExceptionStackContext(self._handle_exception):\n            super(AsyncTestCase, self).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\n    def stop(self, _arg=None, **kwargs):\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        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(self, condition=None, timeout=5):\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.\n\n        If ``condition`` is not None, the `.IOLoop` will be restarted\n        after `stop()` until ``condition()`` returns true.\n        \"\"\"\n        if not self.__stopped:\n            if timeout:\n                def timeout_func():\n                    try:\n                        raise self.failureException(\n                            'Async operation timed out after %s seconds' %\n                            timeout)\n                    except Exception:\n                        self.__failure = sys.exc_info()\n                    self.stop()\n                self.__timeout = self.io_loop.add_timeout(self.io_loop.time() + timeout, timeout_func)\n            while True:\n                self.__running = True\n                self.io_loop.start()\n                if (self.__failure is not None or\n                        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::\n\n        class MyHTTPTest(AsyncHTTPTestCase):\n            def get_app(self):\n                return Application([('/', MyHandler)...])\n\n            def test_homepage(self):\n                # The following two lines are equivalent to\n                #   response = self.fetch('/')\n                # but are shown in full here to demonstrate explicit use\n                # of self.stop and self.wait.\n                self.http_client.fetch(self.get_url('/'), self.stop)\n                response = self.wait()\n                # test contents of response\n    \"\"\"\n    def setUp(self):\n        super(AsyncHTTPTestCase, self).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):\n        return AsyncHTTPClient(io_loop=self.io_loop)\n\n    def get_http_server(self):\n        return HTTPServer(self._app, io_loop=self.io_loop,\n                          **self.get_httpserver_options())\n\n    def get_app(self):\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(self, path, **kwargs):\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 kwargs will be passed directly to\n        `.AsyncHTTPClient.fetch` (and so could be used to pass\n        ``method=\"POST\"``, ``body=\"...\"``, etc).\n        \"\"\"\n        self.http_client.fetch(self.get_url(path), self.stop, **kwargs)\n        return self.wait()\n\n    def get_httpserver_options(self):\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):\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):\n        return 'http'\n\n    def get_url(self, path):\n        \"\"\"Returns an absolute url for the given path on the test server.\"\"\"\n        return '%s://localhost:%s%s' % (self.get_protocol(),\n                                        self.get_http_port(), path)\n\n    def tearDown(self):\n        self.http_server.stop()\n        if (not IOLoop.initialized() or\n                self.http_client.io_loop is not IOLoop.instance()):\n            self.http_client.close()\n        super(AsyncHTTPTestCase, self).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    def get_http_client(self):\n        # Some versions of libcurl have deadlock bugs with ssl,\n        # so always run these tests with SimpleAsyncHTTPClient.\n        return SimpleAsyncHTTPClient(io_loop=self.io_loop, force_instance=True,\n                                     defaults=dict(validate_cert=False))\n\n    def get_httpserver_options(self):\n        return dict(ssl_options=self.get_ssl_options())\n\n    def get_ssl_options(self):\n        \"\"\"May be overridden by subclasses to select SSL options.\n\n        By default includes a self-signed testing certificate.\n        \"\"\"\n        # Testing keys were generated with:\n        # openssl req -new -keyout tornado/test/test.key -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    def get_protocol(self):\n        return 'https'\n\n\ndef gen_test(f):\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 gen.Task(self.fetch('/'))\n\n    \"\"\"\n    f = gen.coroutine(f)\n\n    @functools.wraps(f)\n    def wrapper(self):\n        return self.io_loop.run_sync(functools.partial(f, self), timeout=5)\n    return wrapper\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\n\n\nclass LogTrapTestCase(unittest.TestCase):\n    \"\"\"A test case that captures and discards all logging output\n    if the test passes.\n\n    Some libraries can produce a lot of logging output even when\n    the test succeeds, so this class can be useful to minimize the noise.\n    Simply use it as a base class for your test case.  It is safe to combine\n    with AsyncTestCase via multiple inheritance\n    (``class MyTestCase(AsyncHTTPTestCase, LogTrapTestCase):``)\n\n    This class assumes that only one log handler is configured and\n    that it is a `~logging.StreamHandler`.  This is true for both\n    `logging.basicConfig` and the \"pretty logging\" configured by\n    `tornado.options`.  It is not compatible with other log buffering\n    mechanisms, such as those provided by some test runners.\n    \"\"\"\n    def run(self, result=None):\n        logger = logging.getLogger()\n        if not logger.handlers:\n            logging.basicConfig()\n        handler = logger.handlers[0]\n        if (len(logger.handlers) > 1 or\n                not isinstance(handler, logging.StreamHandler)):\n            # Logging has been configured in a way we don't recognize,\n            # so just leave it alone.\n            super(LogTrapTestCase, self).run(result)\n            return\n        old_stream = handler.stream\n        try:\n            handler.stream = StringIO()\n            gen_log.info(\"RUNNING TEST: \" + str(self))\n            old_error_count = len(result.failures) + len(result.errors)\n            super(LogTrapTestCase, self).run(result)\n            new_error_count = len(result.failures) + len(result.errors)\n            if new_error_count != old_error_count:\n                old_stream.write(handler.stream.getvalue())\n        finally:\n            handler.stream = old_stream\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    Usage::\n\n        with ExpectLog('tornado.application', \"Uncaught exception\"):\n            error_response = self.fetch(\"/some_page\")\n    \"\"\"\n    def __init__(self, logger, regex, required=True):\n        \"\"\"Constructs an ExpectLog context manager.\n\n        :param logger: Logger object (or name of logger) to watch.  Pass\n            an empty string to watch the root logger.\n        :param regex: Regular expression to match.  Any log entries on\n            the specified logger that match this regex will be suppressed.\n        :param required: If true, an exeption will be raised if the end of\n            the ``with`` statement is reached without matching any log entries.\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        self.matched = False\n\n    def filter(self, record):\n        message = record.getMessage()\n        if self.regex.match(message):\n            self.matched = True\n            return False\n        return True\n\n    def __enter__(self):\n        self.logger.addFilter(self)\n\n    def __exit__(self, typ, value, tb):\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\n\ndef main(**kwargs):\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.\n\n    The easiest way to run a test is via the command line::\n\n        python -m tornado.testing tornado.test.stack_context_test\n\n    See the standard library unittest module for ways in which tests can\n    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.stack_context_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    from tornado.options import define, options, parse_command_line\n\n    define('exception_on_interrupt', type=bool, default=True,\n           help=(\"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    # 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    try:\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)\n        else:\n            unittest.main(defaultTest=\"all\", argv=argv, **kwargs)\n    except SystemExit as e:\n        if e.code == 0:\n            gen_log.info('PASS')\n        else:\n            gen_log.error('FAIL')\n        raise\n\nif __name__ == '__main__':\n    main()\n",
            "new_file_content": "#!/usr/bin/env python\n\"\"\"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` and `LogTrapTestCase`: 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\nfrom __future__ import absolute_import, division, print_function, with_statement\n\ntry:\n    from tornado import gen\n    from tornado.httpclient import AsyncHTTPClient\n    from tornado.httpserver import HTTPServer\n    from tornado.simple_httpclient import SimpleAsyncHTTPClient\n    from tornado.ioloop import IOLoop\n    from tornado import netutil\nexcept ImportError:\n    # These modules are not importable on app engine.  Parts of this module\n    # won't work, but e.g. LogTrapTestCase and main() will.\n    AsyncHTTPClient = None\n    gen = None\n    HTTPServer = None\n    IOLoop = None\n    netutil = None\n    SimpleAsyncHTTPClient = None\nfrom tornado.log import gen_log\nfrom tornado.stack_context import ExceptionStackContext\nfrom tornado.util import raise_exc_info, basestring_type\nimport functools\nimport inspect\nimport logging\nimport os\nimport re\nimport signal\nimport socket\nimport sys\n\ntry:\n    from cStringIO import StringIO  # py2\nexcept ImportError:\n    from io import StringIO  # py3\n\n# Tornado's own test suite requires the updated unittest module\n# (either py27+ or unittest2) so tornado.test.util enforces\n# this requirement, but for other users of tornado.testing we want\n# to allow the older version if unitest2 is not available.\ntry:\n    import unittest2 as unittest\nexcept ImportError:\n    import unittest\n\n_next_port = 10000\n\n\ndef get_unused_port():\n    \"\"\"Returns a (hopefully) unused port number.\n\n    This function does not guarantee that the port it returns is available,\n    only that a series of get_unused_port calls in a single process return\n    distinct ports.\n\n    **Deprecated**.  Use bind_unused_port instead, which is guaranteed\n    to find an unused port.\n    \"\"\"\n    global _next_port\n    port = _next_port\n    _next_port = _next_port + 1\n    return port\n\n\ndef bind_unused_port():\n    \"\"\"Binds a server socket to an available port on localhost.\n\n    Returns a tuple (socket, port).\n    \"\"\"\n    [sock] = netutil.bind_sockets(0, 'localhost', family=socket.AF_INET)\n    port = sock.getsockname()[1]\n    return sock, port\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 class provides\n    the `stop()` and `wait()` methods for this purpose.  The test\n    method itself must call ``self.wait()``, and asynchronous\n    callbacks should call ``self.stop()`` to signal completion.\n    Alternately, the `gen_test` decorator can be used to use yield points\n    from the `tornado.gen` module.\n\n    By default, a new `.IOLoop` is constructed for each test and is available\n    as ``self.io_loop``.  This `.IOLoop` should be used in the construction of\n    HTTP clients/servers, etc.  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 argument passing between self.stop and self.wait.\n        class MyTestCase(AsyncTestCase):\n            def test_http_fetch(self):\n                client = AsyncHTTPClient(self.io_loop)\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        # This test uses an explicit callback-based style.\n        class MyTestCase2(AsyncTestCase):\n            def test_http_fetch(self):\n                client = AsyncHTTPClient(self.io_loop)\n                client.fetch(\"http://www.tornadoweb.org/\", self.handle_fetch)\n                self.wait()\n\n            def handle_fetch(self, response):\n                # Test contents of response (failures and exceptions here\n                # will cause self.wait() to throw an exception and end the\n                # test).\n                # Exceptions thrown here are magically propagated to\n                # self.wait() in test_http_fetch() via stack_context.\n                self.assertIn(\"FriendFeed\", response.body)\n                self.stop()\n    \"\"\"\n    def __init__(self, *args, **kwargs):\n        super(AsyncTestCase, self).__init__(*args, **kwargs)\n        self.__stopped = False\n        self.__running = False\n        self.__failure = None\n        self.__stop_args = None\n        self.__timeout = None\n\n    def setUp(self):\n        super(AsyncTestCase, self).setUp()\n        self.io_loop = self.get_new_ioloop()\n        self.io_loop.make_current()\n\n    def tearDown(self):\n        self.io_loop.clear_current()\n        if (not IOLoop.initialized() or\n                self.io_loop is not IOLoop.instance()):\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(AsyncTestCase, self).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):\n        \"\"\"Creates a new `.IOLoop` for this test.  May be overridden in\n        subclasses for tests that require a specific `.IOLoop` (usually\n        the singleton `.IOLoop.instance()`).\n        \"\"\"\n        return IOLoop()\n\n    def _handle_exception(self, typ, value, tb):\n        self.__failure = sys.exc_info()\n        self.stop()\n        return True\n\n    def __rethrow(self):\n        if self.__failure is not None:\n            failure = self.__failure\n            self.__failure = None\n            raise_exc_info(failure)\n\n    def run(self, result=None):\n        with ExceptionStackContext(self._handle_exception):\n            super(AsyncTestCase, self).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\n    def stop(self, _arg=None, **kwargs):\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        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(self, condition=None, timeout=5):\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.\n\n        If ``condition`` is not None, the `.IOLoop` will be restarted\n        after `stop()` until ``condition()`` returns true.\n        \"\"\"\n        if not self.__stopped:\n            if timeout:\n                def timeout_func():\n                    try:\n                        raise self.failureException(\n                            'Async operation timed out after %s seconds' %\n                            timeout)\n                    except Exception:\n                        self.__failure = sys.exc_info()\n                    self.stop()\n                self.__timeout = self.io_loop.add_timeout(self.io_loop.time() + timeout, timeout_func)\n            while True:\n                self.__running = True\n                self.io_loop.start()\n                if (self.__failure is not None or\n                        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::\n\n        class MyHTTPTest(AsyncHTTPTestCase):\n            def get_app(self):\n                return Application([('/', MyHandler)...])\n\n            def test_homepage(self):\n                # The following two lines are equivalent to\n                #   response = self.fetch('/')\n                # but are shown in full here to demonstrate explicit use\n                # of self.stop and self.wait.\n                self.http_client.fetch(self.get_url('/'), self.stop)\n                response = self.wait()\n                # test contents of response\n    \"\"\"\n    def setUp(self):\n        super(AsyncHTTPTestCase, self).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):\n        return AsyncHTTPClient(io_loop=self.io_loop)\n\n    def get_http_server(self):\n        return HTTPServer(self._app, io_loop=self.io_loop,\n                          **self.get_httpserver_options())\n\n    def get_app(self):\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(self, path, **kwargs):\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 kwargs will be passed directly to\n        `.AsyncHTTPClient.fetch` (and so could be used to pass\n        ``method=\"POST\"``, ``body=\"...\"``, etc).\n        \"\"\"\n        self.http_client.fetch(self.get_url(path), self.stop, **kwargs)\n        return self.wait()\n\n    def get_httpserver_options(self):\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):\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):\n        return 'http'\n\n    def get_url(self, path):\n        \"\"\"Returns an absolute url for the given path on the test server.\"\"\"\n        return '%s://localhost:%s%s' % (self.get_protocol(),\n                                        self.get_http_port(), path)\n\n    def tearDown(self):\n        self.http_server.stop()\n        if (not IOLoop.initialized() or\n                self.http_client.io_loop is not IOLoop.instance()):\n            self.http_client.close()\n        super(AsyncHTTPTestCase, self).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    def get_http_client(self):\n        # Some versions of libcurl have deadlock bugs with ssl,\n        # so always run these tests with SimpleAsyncHTTPClient.\n        return SimpleAsyncHTTPClient(io_loop=self.io_loop, force_instance=True,\n                                     defaults=dict(validate_cert=False))\n\n    def get_httpserver_options(self):\n        return dict(ssl_options=self.get_ssl_options())\n\n    def get_ssl_options(self):\n        \"\"\"May be overridden by subclasses to select SSL options.\n\n        By default includes a self-signed testing certificate.\n        \"\"\"\n        # Testing keys were generated with:\n        # openssl req -new -keyout tornado/test/test.key -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    def get_protocol(self):\n        return 'https'\n\n\ndef gen_test(timeout=None):\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 gen.Task(self.fetch('/'))\n\n    By default, ``@gen_test`` times out after 5 seconds. The timeout may be\n    overridden globally with the TIMEOUT environment variable, or for each\n    test with the ``timeout`` parameter:\n\n        class MyTest(AsyncHTTPTestCase):\n            @gen_test(timeout=10)\n            def test_something_slow(self):\n                response = yield gen.Task(self.fetch('/'))\n\n    If both the environment variable and the parameter are set, ``gen_test``\n    uses the maximum of the two.\n    \"\"\"\n    try:\n        env_timeout = float(os.environ.get('TIMEOUT'))\n    except (ValueError, TypeError):\n        env_timeout = None\n\n    def wrap(f):\n        f = gen.coroutine(f)\n\n        @functools.wraps(f)\n        def wrapper(self):\n            return self.io_loop.run_sync(\n                functools.partial(f, self), timeout=timeout)\n        return wrapper\n\n    if inspect.isfunction(timeout):\n        # Used like:\n        #     @gen_test\n        #     def f(self):\n        #         pass\n        # The 'timeout' parameter is actually the test function.\n        f = timeout\n        timeout = env_timeout or 5\n        return wrap(f)\n    else:\n        # Used like @gen_test(timeout=10) or @gen_test(10).\n        if env_timeout is not None:\n            timeout = max(float(timeout), env_timeout)\n        else:\n            timeout = float(timeout)\n\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\n\n\nclass LogTrapTestCase(unittest.TestCase):\n    \"\"\"A test case that captures and discards all logging output\n    if the test passes.\n\n    Some libraries can produce a lot of logging output even when\n    the test succeeds, so this class can be useful to minimize the noise.\n    Simply use it as a base class for your test case.  It is safe to combine\n    with AsyncTestCase via multiple inheritance\n    (``class MyTestCase(AsyncHTTPTestCase, LogTrapTestCase):``)\n\n    This class assumes that only one log handler is configured and\n    that it is a `~logging.StreamHandler`.  This is true for both\n    `logging.basicConfig` and the \"pretty logging\" configured by\n    `tornado.options`.  It is not compatible with other log buffering\n    mechanisms, such as those provided by some test runners.\n    \"\"\"\n    def run(self, result=None):\n        logger = logging.getLogger()\n        if not logger.handlers:\n            logging.basicConfig()\n        handler = logger.handlers[0]\n        if (len(logger.handlers) > 1 or\n                not isinstance(handler, logging.StreamHandler)):\n            # Logging has been configured in a way we don't recognize,\n            # so just leave it alone.\n            super(LogTrapTestCase, self).run(result)\n            return\n        old_stream = handler.stream\n        try:\n            handler.stream = StringIO()\n            gen_log.info(\"RUNNING TEST: \" + str(self))\n            old_error_count = len(result.failures) + len(result.errors)\n            super(LogTrapTestCase, self).run(result)\n            new_error_count = len(result.failures) + len(result.errors)\n            if new_error_count != old_error_count:\n                old_stream.write(handler.stream.getvalue())\n        finally:\n            handler.stream = old_stream\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    Usage::\n\n        with ExpectLog('tornado.application', \"Uncaught exception\"):\n            error_response = self.fetch(\"/some_page\")\n    \"\"\"\n    def __init__(self, logger, regex, required=True):\n        \"\"\"Constructs an ExpectLog context manager.\n\n        :param logger: Logger object (or name of logger) to watch.  Pass\n            an empty string to watch the root logger.\n        :param regex: Regular expression to match.  Any log entries on\n            the specified logger that match this regex will be suppressed.\n        :param required: If true, an exeption will be raised if the end of\n            the ``with`` statement is reached without matching any log entries.\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        self.matched = False\n\n    def filter(self, record):\n        message = record.getMessage()\n        if self.regex.match(message):\n            self.matched = True\n            return False\n        return True\n\n    def __enter__(self):\n        self.logger.addFilter(self)\n\n    def __exit__(self, typ, value, tb):\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\n\ndef main(**kwargs):\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.\n\n    The easiest way to run a test is via the command line::\n\n        python -m tornado.testing tornado.test.stack_context_test\n\n    See the standard library unittest module for ways in which tests can\n    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.stack_context_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    from tornado.options import define, options, parse_command_line\n\n    define('exception_on_interrupt', type=bool, default=True,\n           help=(\"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    # 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    try:\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)\n        else:\n            unittest.main(defaultTest=\"all\", argv=argv, **kwargs)\n    except SystemExit as e:\n        if e.code == 0:\n            gen_log.info('PASS')\n        else:\n            gen_log.error('FAIL')\n        raise\n\nif __name__ == '__main__':\n    main()\n",
            "header": {
                "file": {
                    "path": "tornado/testing.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "61c84306",
                "new_commit_hash": "1330ea20",
                "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": 32,
                            "length": 6
                        },
                        "new_range": {
                            "start": 32,
                            "length": 7
                        },
                        "section": "from tornado.log import gen_log"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "from tornado.stack_context import ExceptionStackContext",
                                "type": "context"
                            },
                            {
                                "content": "from tornado.util import raise_exc_info, basestring_type",
                                "type": "context"
                            },
                            {
                                "content": "import functools",
                                "type": "context"
                            },
                            {
                                "content": "import inspect",
                                "type": "added"
                            },
                            {
                                "content": "import logging",
                                "type": "context"
                            },
                            {
                                "content": "import os",
                                "type": "context"
                            },
                            {
                                "content": "import re",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [],
                    "added_entities": [
                        {
                            "file_name": "tornado/testing.py",
                            "type": "import",
                            "name": "inspect",
                            "content": "import inspect",
                            "ast_type_str": "Import",
                            "start_lineno": 35,
                            "end_lineno": 35,
                            "parent": null
                        }
                    ],
                    "deleted_entities": []
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 354,
                            "length": 7
                        },
                        "new_range": {
                            "start": 355,
                            "length": 7
                        },
                        "section": "class AsyncHTTPSTestCase(AsyncHTTPTestCase):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        return 'https'",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "def gen_test(f):",
                                "type": "deleted"
                            },
                            {
                                "content": "def gen_test(timeout=None):",
                                "type": "added"
                            },
                            {
                                "content": "    \"\"\"Testing equivalent of ``@gen.coroutine``, to be applied to test methods.",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    ``@gen.coroutine`` cannot be used on tests because the `.IOLoop` is not",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/testing.py",
                            "type": "function",
                            "name": "gen_test",
                            "content": "def gen_test(f):\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 gen.Task(self.fetch('/'))\n\n    \"\"\"\n    f = gen.coroutine(f)\n\n    @functools.wraps(f)\n    def wrapper(self):\n        return self.io_loop.run_sync(functools.partial(f, self), timeout=5)\n    return wrapper",
                            "ast_type_str": "FunctionDef",
                            "start_lineno": 357,
                            "end_lineno": 377,
                            "parent": null
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 368,
                            "length": 13
                        },
                        "new_range": {
                            "start": 369,
                            "length": 49
                        },
                        "section": "def gen_test(f):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "            def test_something(self):",
                                "type": "context"
                            },
                            {
                                "content": "                response = yield gen.Task(self.fetch('/'))",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    By default, ``@gen_test`` times out after 5 seconds. The timeout may be",
                                "type": "added"
                            },
                            {
                                "content": "    overridden globally with the TIMEOUT environment variable, or for each",
                                "type": "added"
                            },
                            {
                                "content": "    test with the ``timeout`` parameter:",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "        class MyTest(AsyncHTTPTestCase):",
                                "type": "added"
                            },
                            {
                                "content": "            @gen_test(timeout=10)",
                                "type": "added"
                            },
                            {
                                "content": "            def test_something_slow(self):",
                                "type": "added"
                            },
                            {
                                "content": "                response = yield gen.Task(self.fetch('/'))",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "    If both the environment variable and the parameter are set, ``gen_test``",
                                "type": "added"
                            },
                            {
                                "content": "    uses the maximum of the two.",
                                "type": "added"
                            },
                            {
                                "content": "    \"\"\"",
                                "type": "context"
                            },
                            {
                                "content": "    f = gen.coroutine(f)",
                                "type": "deleted"
                            },
                            {
                                "content": "    try:",
                                "type": "added"
                            },
                            {
                                "content": "        env_timeout = float(os.environ.get('TIMEOUT'))",
                                "type": "added"
                            },
                            {
                                "content": "    except (ValueError, TypeError):",
                                "type": "added"
                            },
                            {
                                "content": "        env_timeout = None",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "    def wrap(f):",
                                "type": "added"
                            },
                            {
                                "content": "        f = gen.coroutine(f)",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "        @functools.wraps(f)",
                                "type": "added"
                            },
                            {
                                "content": "        def wrapper(self):",
                                "type": "added"
                            },
                            {
                                "content": "            return self.io_loop.run_sync(",
                                "type": "added"
                            },
                            {
                                "content": "                functools.partial(f, self), timeout=timeout)",
                                "type": "added"
                            },
                            {
                                "content": "        return wrapper",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "    if inspect.isfunction(timeout):",
                                "type": "added"
                            },
                            {
                                "content": "        # Used like:",
                                "type": "added"
                            },
                            {
                                "content": "        #     @gen_test",
                                "type": "added"
                            },
                            {
                                "content": "        #     def f(self):",
                                "type": "added"
                            },
                            {
                                "content": "        #         pass",
                                "type": "added"
                            },
                            {
                                "content": "        # The 'timeout' parameter is actually the test function.",
                                "type": "added"
                            },
                            {
                                "content": "        f = timeout",
                                "type": "added"
                            },
                            {
                                "content": "        timeout = env_timeout or 5",
                                "type": "added"
                            },
                            {
                                "content": "        return wrap(f)",
                                "type": "added"
                            },
                            {
                                "content": "    else:",
                                "type": "added"
                            },
                            {
                                "content": "        # Used like @gen_test(timeout=10) or @gen_test(10).",
                                "type": "added"
                            },
                            {
                                "content": "        if env_timeout is not None:",
                                "type": "added"
                            },
                            {
                                "content": "            timeout = max(float(timeout), env_timeout)",
                                "type": "added"
                            },
                            {
                                "content": "        else:",
                                "type": "added"
                            },
                            {
                                "content": "            timeout = float(timeout)",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    @functools.wraps(f)",
                                "type": "deleted"
                            },
                            {
                                "content": "    def wrapper(self):",
                                "type": "deleted"
                            },
                            {
                                "content": "        return self.io_loop.run_sync(functools.partial(f, self), timeout=5)",
                                "type": "deleted"
                            },
                            {
                                "content": "    return wrapper",
                                "type": "deleted"
                            },
                            {
                                "content": "        return wrap",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "# Without this attribute, nosetests will try to run gen_test as a test",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "tornado/testing.py",
                            "type": "function",
                            "name": "gen_test",
                            "content": "def gen_test(f):\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 gen.Task(self.fetch('/'))\n\n    \"\"\"\n    f = gen.coroutine(f)\n\n    @functools.wraps(f)\n    def wrapper(self):\n        return self.io_loop.run_sync(functools.partial(f, self), timeout=5)\n    return wrapper",
                            "ast_type_str": "FunctionDef",
                            "start_lineno": 357,
                            "end_lineno": 377,
                            "parent": null
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                }
            ]
        }
    ],
    "old_commit_hash": "cab39beba263e68647d54d68d18cab4f52c7a802^",
    "new_commit_hash": "cab39beba263e68647d54d68d18cab4f52c7a802",
    "commit_message": "Make gen_test's timeout configurable",
    "commit_date": "2013-04-11T15:17:26-04:00",
    "metadata": {}
}
{
    "repo_name": "tornado",
    "new_commit_hash": "cab39beba263e68647d54d68d18cab4f52c7a802",
    "test_file_codes": [
        "#!/usr/bin/env python\n\nfrom __future__ import absolute_import, division, print_function, with_statement\n\nfrom tornado import gen, ioloop\nfrom tornado.testing import AsyncTestCase, gen_test\nfrom tornado.test.util import unittest\n\nimport functools\nimport os\n\n\nclass AsyncTestCaseTest(AsyncTestCase):\n    def test_exception_in_callback(self):\n        self.io_loop.add_callback(lambda: 1 / 0)\n        try:\n            self.wait()\n            self.fail(\"did not get expected exception\")\n        except ZeroDivisionError:\n            pass\n\n    def test_subsequent_wait_calls(self):\n        \"\"\"\n        This test makes sure that a second call to wait()\n        clears the first timeout.\n        \"\"\"\n        self.io_loop.add_timeout(self.io_loop.time() + 0.01, self.stop)\n        self.wait(timeout=0.02)\n        self.io_loop.add_timeout(self.io_loop.time() + 0.03, self.stop)\n        self.wait(timeout=0.15)\n\n\nclass SetUpTearDownTest(unittest.TestCase):\n    def test_set_up_tear_down(self):\n        \"\"\"\n        This test makes sure that AsyncTestCase calls super methods for\n        setUp and tearDown.\n\n        InheritBoth is a subclass of both AsyncTestCase and\n        SetUpTearDown, with the ordering so that the super of\n        AsyncTestCase will be SetUpTearDown.\n        \"\"\"\n        events = []\n        result = unittest.TestResult()\n\n        class SetUpTearDown(unittest.TestCase):\n            def setUp(self):\n                events.append('setUp')\n\n            def tearDown(self):\n                events.append('tearDown')\n\n        class InheritBoth(AsyncTestCase, SetUpTearDown):\n            def test(self):\n                events.append('test')\n\n        InheritBoth('test').run(result)\n        expected = ['setUp', 'test', 'tearDown']\n        self.assertEqual(expected, events)\n\n\nclass GenTest(AsyncTestCase):\n    def setUp(self):\n        super(GenTest, self).setUp()\n        self.finished = False\n\n    def tearDown(self):\n        self.assertTrue(self.finished)\n        super(GenTest, self).tearDown()\n\n    @gen_test\n    def test_sync(self):\n        self.finished = True\n\n    @gen_test\n    def test_async(self):\n        yield gen.Task(self.io_loop.add_callback)\n        self.finished = True\n\n    def test_timeout(self):\n        # Set a short timeout and exceed it.\n        @gen_test(timeout=0.1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)\n\n        with self.assertRaises(ioloop.TimeoutError):\n            test(self)\n\n        self.finished = True\n\n    def test_no_timeout(self):\n        # A test that does not exceed its timeout should succeed.\n        @gen_test(timeout=1)\n        def test(self):\n            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 0.1)\n\n        test(self)\n        self.finished = True\n\n    def test_timeout_environment_variable(self):\n        time = self.io_loop.time\n        add_timeout = self.io_loop.add_timeout\n        old_timeout = os.environ.get('TIMEOUT')\n        try:\n            os.environ['TIMEOUT'] = '0.1'\n\n            @gen_test(timeout=0.5)\n            def test_long_timeout(self):\n                yield gen.Task(add_timeout, time() + 0.25)\n\n            # Uses provided timeout of 0.5 seconds, doesn't time out.\n            self.io_loop.run_sync(\n                functools.partial(test_long_timeout, self))\n\n            @gen_test(timeout=0.01)\n            def test_short_timeout(self):\n                yield gen.Task(add_timeout, time() + 1)\n\n            # Uses environment TIMEOUT of 0.1, times out.\n            with self.assertRaises(ioloop.TimeoutError):\n                test_short_timeout(self)\n\n            self.finished = True\n        finally:\n            if old_timeout is None:\n                del os.environ['TIMEOUT']\n            else:\n                os.environ['TIMEOUT'] = old_timeout\n\nif __name__ == '__main__':\n    unittest.main()\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 703ms\nPrepared 1 package in 827ms\nInstalled 1 package in 0.79ms\n + tornado==3.0.1 (from file:///home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_cab39beba263e68647d54d68d18cab4f52c7a802)\nResolved 29 packages in 10ms\nInstalled 29 packages in 35ms\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": 0,
    "new_commit_res_stdout": "============================= test session starts =============================\nplatform linux -- Python 3.9.21\nrootdir: /home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_cab39beba263e68647d54d68d18cab4f52c7a802\n........\ncollected 8 items\n\n==================== short test summary info ====================\nPASSED r2e_tests.test_1::AsyncTestCaseTest::test_exception_in_callback\nPASSED r2e_tests.test_1::AsyncTestCaseTest::test_subsequent_wait_calls\nPASSED r2e_tests.test_1::GenTest::test_async\nPASSED r2e_tests.test_1::GenTest::test_no_timeout\nPASSED r2e_tests.test_1::GenTest::test_sync\nPASSED r2e_tests.test_1::GenTest::test_timeout\nPASSED r2e_tests.test_1::GenTest::test_timeout_environment_variable\nPASSED r2e_tests.test_1::SetUpTearDownTest::test_set_up_tear_down\n=================== 8 passed in 0.50s ===================\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_cab39beba263e68647d54d68d18cab4f52c7a802\n........\ncollected 11 items\n\n=================================== FAILURES ===================================\n__________ GenTest.test_no_timeout __________\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 595, in run\n    self._callTearDown()\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 553, in _callTearDown\n    self.tearDown()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_cab39beba263e68647d54d68d18cab4f52c7a802/r2e_tests/test_1.py\", line 68, in tearDown\n    self.assertTrue(self.finished)\nAssertionError: False is not true\n__________ GenTest.test_timeout __________\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 595, in run\n    self._callTearDown()\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 553, in _callTearDown\n    self.tearDown()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_cab39beba263e68647d54d68d18cab4f52c7a802/r2e_tests/test_1.py\", line 68, in tearDown\n    self.assertTrue(self.finished)\nAssertionError: False is not true\n__________ GenTest.test_timeout_environment_variable __________\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 595, in run\n    self._callTearDown()\n  File \"/home/gcpuser/.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py\", line 553, in _callTearDown\n    self.tearDown()\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/tornado_cab39beba263e68647d54d68d18cab4f52c7a802/r2e_tests/test_1.py\", line 68, in tearDown\n    self.assertTrue(self.finished)\nAssertionError: False is not true\n\n=================================== ERRORS =====================================\n__________ GenTest.test_no_timeout __________\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_cab39beba263e68647d54d68d18cab4f52c7a802/r2e_tests/test_1.py\", line 93, in test_no_timeout\n    @gen_test(timeout=1)\nTypeError: gen_test() got an unexpected keyword argument 'timeout'\n__________ GenTest.test_timeout __________\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_cab39beba263e68647d54d68d18cab4f52c7a802/r2e_tests/test_1.py\", line 82, in test_timeout\n    @gen_test(timeout=0.1)\nTypeError: gen_test() got an unexpected keyword argument 'timeout'\n__________ GenTest.test_timeout_environment_variable __________\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_cab39beba263e68647d54d68d18cab4f52c7a802/r2e_tests/test_1.py\", line 107, in test_timeout_environment_variable\n    @gen_test(timeout=0.5)\nTypeError: gen_test() got an unexpected keyword argument 'timeout'\n\n==================== short test summary info ====================\nPASSED r2e_tests.test_1::AsyncTestCaseTest::test_exception_in_callback\nPASSED r2e_tests.test_1::AsyncTestCaseTest::test_subsequent_wait_calls\nPASSED r2e_tests.test_1::GenTest::test_async\nERROR r2e_tests.test_1::GenTest::test_no_timeout\nFAILED r2e_tests.test_1::GenTest::test_no_timeout\nPASSED r2e_tests.test_1::GenTest::test_sync\nERROR r2e_tests.test_1::GenTest::test_timeout\nFAILED r2e_tests.test_1::GenTest::test_timeout\nERROR r2e_tests.test_1::GenTest::test_timeout_environment_variable\nFAILED r2e_tests.test_1::GenTest::test_timeout_environment_variable\nPASSED r2e_tests.test_1::SetUpTearDownTest::test_set_up_tear_down\n=================== 3 failed, 3 errors, 5 passed in 0.04s ===================\n",
    "old_commit_res_stderr": ""
}
tornado/test/testing_test.py tornado/testing.py
We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
**[ISSUE]**

**Title:** `gen_test` Decorator Does Not Accept `timeout` Parameter, Causing TypeError

**Description:**
When attempting to use the `gen_test` decorator with the `timeout` parameter to specify a custom timeout for asynchronous tests, a `TypeError` is raised. This prevents tests that require longer execution times from running successfully.

**Example Code:**
```python
from tornado import gen, ioloop
from tornado.testing import AsyncTestCase, gen_test

class MyAsyncTests(AsyncTestCase):

    @gen_test(timeout=10)
    def test_long_operation(self):
        # Simulate a long-running asynchronous operation
        yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 15)
```

**Expected Behavior:**
The `timeout` parameter should allow configuring the maximum duration the test can run before timing out. In this example, the test should allow up to 10 seconds for `test_long_operation` to complete.

**Actual Behavior:**
Running the test results in the following error:
```
TypeError: gen_test() got an unexpected keyword argument 'timeout'
```
This error indicates that the `gen_test` decorator does not recognize the `timeout` parameter, causing the test to fail immediately instead of adhering to the specified timeout.

**[/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
#!/usr/bin/env python
"""Support classes for automated testing.

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

* `ExpectLog` and `LogTrapTestCase`: 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.
"""

from __future__ import absolute_import, division, print_function, with_statement

try:
    from tornado import gen
    from tornado.httpclient import AsyncHTTPClient
    from tornado.httpserver import HTTPServer
    from tornado.simple_httpclient import SimpleAsyncHTTPClient
    from tornado.ioloop import IOLoop
    from tornado import netutil
except ImportError:
    # These modules are not importable on app engine.  Parts of this module
    # won't work, but e.g. LogTrapTestCase and main() will.
    AsyncHTTPClient = None
    gen = None
    HTTPServer = None
    IOLoop = None
    netutil = None
    SimpleAsyncHTTPClient = None
from tornado.log import gen_log
from tornado.stack_context import ExceptionStackContext
from tornado.util import raise_exc_info, basestring_type
import functools
import logging
import os
import re
import signal
import socket
import sys

try:
    from cStringIO import StringIO  # py2
except ImportError:
    from io import StringIO  # py3

# Tornado's own test suite requires the updated unittest module
# (either py27+ or unittest2) so tornado.test.util enforces
# this requirement, but for other users of tornado.testing we want
# to allow the older version if unitest2 is not available.
try:
    import unittest2 as unittest
except ImportError:
    import unittest

_next_port = 10000


def get_unused_port():
    """Returns a (hopefully) unused port number.

    This function does not guarantee that the port it returns is available,
    only that a series of get_unused_port calls in a single process return
    distinct ports.

    **Deprecated**.  Use bind_unused_port instead, which is guaranteed
    to find an unused port.
    """
    global _next_port
    port = _next_port
    _next_port = _next_port + 1
    return port


def bind_unused_port():
    """Binds a server socket to an available port on localhost.

    Returns a tuple (socket, port).
    """
    [sock] = netutil.bind_sockets(0, 'localhost', family=socket.AF_INET)
    port = sock.getsockname()[1]
    return sock, port


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 class provides
    the `stop()` and `wait()` methods for this purpose.  The test
    method itself must call ``self.wait()``, and asynchronous
    callbacks should call ``self.stop()`` to signal completion.
    Alternately, the `gen_test` decorator can be used to use yield points
    from the `tornado.gen` module.

    By default, a new `.IOLoop` is constructed for each test and is available
    as ``self.io_loop``.  This `.IOLoop` should be used in the construction of
    HTTP clients/servers, etc.  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 argument passing between self.stop and self.wait.
        class MyTestCase(AsyncTestCase):
            def test_http_fetch(self):
                client = AsyncHTTPClient(self.io_loop)
                client.fetch("http://www.tornadoweb.org/", self.stop)
                response = self.wait()
                # Test contents of response
                self.assertIn("FriendFeed", response.body)

        # This test uses an explicit callback-based style.
        class MyTestCase2(AsyncTestCase):
            def test_http_fetch(self):
                client = AsyncHTTPClient(self.io_loop)
                client.fetch("http://www.tornadoweb.org/", self.handle_fetch)
                self.wait()

            def handle_fetch(self, response):
                # Test contents of response (failures and exceptions here
                # will cause self.wait() to throw an exception and end the
                # test).
                # Exceptions thrown here are magically propagated to
                # self.wait() in test_http_fetch() via stack_context.
                self.assertIn("FriendFeed", response.body)
                self.stop()
    """
    def __init__(self, *args, **kwargs):
        super(AsyncTestCase, self).__init__(*args, **kwargs)
        self.__stopped = False
        self.__running = False
        self.__failure = None
        self.__stop_args = None
        self.__timeout = None

    def setUp(self):
        super(AsyncTestCase, self).setUp()
        self.io_loop = self.get_new_ioloop()
        self.io_loop.make_current()

    def tearDown(self):
        self.io_loop.clear_current()
        if (not IOLoop.initialized() or
                self.io_loop is not IOLoop.instance()):
            # 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)
        super(AsyncTestCase, self).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):
        """Creates a new `.IOLoop` for this test.  May be overridden in
        subclasses for tests that require a specific `.IOLoop` (usually
        the singleton `.IOLoop.instance()`).
        """
        return IOLoop()

    def _handle_exception(self, typ, value, tb):
        self.__failure = sys.exc_info()
        self.stop()
        return True

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

    def run(self, result=None):
        with ExceptionStackContext(self._handle_exception):
            super(AsyncTestCase, self).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()

    def stop(self, _arg=None, **kwargs):
        """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()`.
        """
        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=None, timeout=5):
        """Runs the `.IOLoop` until stop is called or timeout has passed.

        In the event of a timeout, an exception will be thrown.

        If ``condition`` is not None, the `.IOLoop` will be restarted
        after `stop()` until ``condition()`` returns true.
        """
        if not self.__stopped:
            if timeout:
                def timeout_func():
                    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::

        class MyHTTPTest(AsyncHTTPTestCase):
            def get_app(self):
                return Application([('/', MyHandler)...])

            def test_homepage(self):
                # The following two lines are equivalent to
                #   response = self.fetch('/')
                # but are shown in full here to demonstrate explicit use
                # of self.stop and self.wait.
                self.http_client.fetch(self.get_url('/'), self.stop)
                response = self.wait()
                # test contents of response
    """
    def setUp(self):
        super(AsyncHTTPTestCase, self).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):
        return AsyncHTTPClient(io_loop=self.io_loop)

    def get_http_server(self):
        return HTTPServer(self._app, io_loop=self.io_loop,
                          **self.get_httpserver_options())

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

    def fetch(self, path, **kwargs):
        """Convenience method to synchronously fetch a url.

        The given path will be appended to the local server's host and
        port.  Any additional kwargs will be passed directly to
        `.AsyncHTTPClient.fetch` (and so could be used to pass
        ``method="POST"``, ``body="..."``, etc).
        """
        self.http_client.fetch(self.get_url(path), self.stop, **kwargs)
        return self.wait()

    def get_httpserver_options(self):
        """May be overridden by subclasses to return additional
        keyword arguments for the server.
        """
        return {}

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

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

    def get_protocol(self):
        return 'http'

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

    def tearDown(self):
        self.http_server.stop()
        if (not IOLoop.initialized() or
                self.http_client.io_loop is not IOLoop.instance()):
            self.http_client.close()
        super(AsyncHTTPTestCase, self).tearDown()


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

    Interface is generally the same as `AsyncHTTPTestCase`.
    """
    def get_http_client(self):
        # Some versions of libcurl have deadlock bugs with ssl,
        # so always run these tests with SimpleAsyncHTTPClient.
        return SimpleAsyncHTTPClient(io_loop=self.io_loop, force_instance=True,
                                     defaults=dict(validate_cert=False))

    def get_httpserver_options(self):
        return dict(ssl_options=self.get_ssl_options())

    def get_ssl_options(self):
        """May be overridden by subclasses to select SSL options.

        By default includes a self-signed testing certificate.
        """
        # 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):
        return 'https'


def gen_test(f):
    """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 gen.Task(self.fetch('/'))

    """
    f = gen.coroutine(f)

    @functools.wraps(f)
    def wrapper(self):
        return self.io_loop.run_sync(functools.partial(f, self), timeout=5)
    return wrapper


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


class LogTrapTestCase(unittest.TestCase):
    """A test case that captures and discards all logging output
    if the test passes.

    Some libraries can produce a lot of logging output even when
    the test succeeds, so this class can be useful to minimize the noise.
    Simply use it as a base class for your test case.  It is safe to combine
    with AsyncTestCase via multiple inheritance
    (``class MyTestCase(AsyncHTTPTestCase, LogTrapTestCase):``)

    This class assumes that only one log handler is configured and
    that it is a `~logging.StreamHandler`.  This is true for both
    `logging.basicConfig` and the "pretty logging" configured by
    `tornado.options`.  It is not compatible with other log buffering
    mechanisms, such as those provided by some test runners.
    """
    def run(self, result=None):
        logger = logging.getLogger()
        if not logger.handlers:
            logging.basicConfig()
        handler = logger.handlers[0]
        if (len(logger.handlers) > 1 or
                not isinstance(handler, logging.StreamHandler)):
            # Logging has been configured in a way we don't recognize,
            # so just leave it alone.
            super(LogTrapTestCase, self).run(result)
            return
        old_stream = handler.stream
        try:
            handler.stream = StringIO()
            gen_log.info("RUNNING TEST: " + str(self))
            old_error_count = len(result.failures) + len(result.errors)
            super(LogTrapTestCase, self).run(result)
            new_error_count = len(result.failures) + len(result.errors)
            if new_error_count != old_error_count:
                old_stream.write(handler.stream.getvalue())
        finally:
            handler.stream = old_stream


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.*

    Usage::

        with ExpectLog('tornado.application', "Uncaught exception"):
            error_response = self.fetch("/some_page")
    """
    def __init__(self, logger, regex, required=True):
        """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 exeption will be raised if the end of
            the ``with`` statement is reached without matching any log entries.
        """
        if isinstance(logger, basestring_type):
            logger = logging.getLogger(logger)
        self.logger = logger
        self.regex = re.compile(regex)
        self.required = required
        self.matched = False

    def filter(self, record):
        message = record.getMessage()
        if self.regex.match(message):
            self.matched = True
            return False
        return True

    def __enter__(self):
        self.logger.addFilter(self)

    def __exit__(self, typ, value, tb):
        self.logger.removeFilter(self)
        if not typ and self.required and not self.matched:
            raise Exception("did not get expected log message")


def main(**kwargs):
    """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.

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

        python -m tornado.testing tornado.test.stack_context_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.stack_context_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.
    """
    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)
    try:
        # 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)
        else:
            unittest.main(defaultTest="all", argv=argv, **kwargs)
    except SystemExit as e:
        if e.code == 0:
            gen_log.info('PASS')
        else:
            gen_log.error('FAIL')
        raise

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:** `gen_test` Decorator Does Not Accept `timeout` Parameter, Causing TypeError

**Description:**
When attempting to use the `gen_test` decorator with the `timeout` parameter to specify a custom timeout for asynchronous tests, a `TypeError` is raised. This prevents tests that require longer execution times from running successfully.

**Example Code:**
```python
from tornado import gen, ioloop
from tornado.testing import AsyncTestCase, gen_test

class MyAsyncTests(AsyncTestCase):

    @gen_test(timeout=10)
    def test_long_operation(self):
        # Simulate a long-running asynchronous operation
        yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 15)
```

**Expected Behavior:**
The `timeout` parameter should allow configuring the maximum duration the test can run before timing out. In this example, the test should allow up to 10 seconds for `test_long_operation` to complete.

**Actual Behavior:**
Running the test results in the following error:
```
TypeError: gen_test() got an unexpected keyword argument 'timeout'
```
This error indicates that the `gen_test` decorator does not recognize the `timeout` parameter, causing the test to fail immediately instead of adhering to the specified timeout.

**[/ISSUE]**
{
    "AsyncTestCaseTest.test_exception_in_callback": "PASSED",
    "AsyncTestCaseTest.test_subsequent_wait_calls": "PASSED",
    "GenTest.test_async": "PASSED",
    "GenTest.test_no_timeout": "PASSED",
    "GenTest.test_sync": "PASSED",
    "GenTest.test_timeout": "PASSED",
    "GenTest.test_timeout_environment_variable": "PASSED",
    "SetUpTearDownTest.test_set_up_tear_down": "PASSED"
}
diff --git a/tornado/testing.py b/tornado/testing.py
index 61c84306..1330ea20 100644
--- a/tornado/testing.py
+++ b/tornado/testing.py
@@ -32,6 +32,7 @@ from tornado.log import gen_log
 from tornado.stack_context import ExceptionStackContext
 from tornado.util import raise_exc_info, basestring_type
 import functools
+import inspect
 import logging
 import os
 import re
@@ -354,7 +355,7 @@ class AsyncHTTPSTestCase(AsyncHTTPTestCase):
         return 'https'
 
 
-def gen_test(f):
+def gen_test(timeout=None):
     """Testing equivalent of ``@gen.coroutine``, to be applied to test methods.
 
     ``@gen.coroutine`` cannot be used on tests because the `.IOLoop` is not
@@ -368,13 +369,49 @@ def gen_test(f):
             def test_something(self):
                 response = yield gen.Task(self.fetch('/'))
 
+    By default, ``@gen_test`` times out after 5 seconds. The timeout may be
+    overridden globally with the TIMEOUT environment variable, or for each
+    test with the ``timeout`` parameter:
+
+        class MyTest(AsyncHTTPTestCase):
+            @gen_test(timeout=10)
+            def test_something_slow(self):
+                response = yield gen.Task(self.fetch('/'))
+
+    If both the environment variable and the parameter are set, ``gen_test``
+    uses the maximum of the two.
     """
-    f = gen.coroutine(f)
+    try:
+        env_timeout = float(os.environ.get('TIMEOUT'))
+    except (ValueError, TypeError):
+        env_timeout = None
+
+    def wrap(f):
+        f = gen.coroutine(f)
+
+        @functools.wraps(f)
+        def wrapper(self):
+            return self.io_loop.run_sync(
+                functools.partial(f, self), timeout=timeout)
+        return wrapper
+
+    if inspect.isfunction(timeout):
+        # Used like:
+        #     @gen_test
+        #     def f(self):
+        #         pass
+        # The 'timeout' parameter is actually the test function.
+        f = timeout
+        timeout = env_timeout or 5
+        return wrap(f)
+    else:
+        # Used like @gen_test(timeout=10) or @gen_test(10).
+        if env_timeout is not None:
+            timeout = max(float(timeout), env_timeout)
+        else:
+            timeout = float(timeout)
 
-    @functools.wraps(f)
-    def wrapper(self):
-        return self.io_loop.run_sync(functools.partial(f, self), timeout=5)
-    return wrapper
+        return wrap
 
 
 # Without this attribute, nosetests will try to run gen_test as a test
 

{"tornado/testing.py": "#!/usr/bin/env python\n\"\"\"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` and `LogTrapTestCase`: 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\nfrom __future__ import absolute_import, division, print_function, with_statement\n\ntry:\n    from tornado import gen\n    from tornado.httpclient import AsyncHTTPClient\n    from tornado.httpserver import HTTPServer\n    from tornado.simple_httpclient import SimpleAsyncHTTPClient\n    from tornado.ioloop import IOLoop\n    from tornado import netutil\nexcept ImportError:\n    # These modules are not importable on app engine.  Parts of this module\n    # won't work, but e.g. LogTrapTestCase and main() will.\n    AsyncHTTPClient = None\n    gen = None\n    HTTPServer = None\n    IOLoop = None\n    netutil = None\n    SimpleAsyncHTTPClient = None\nfrom tornado.log import gen_log\nfrom tornado.stack_context import ExceptionStackContext\nfrom tornado.util import raise_exc_info, basestring_type\nimport functools\nimport logging\nimport os\nimport re\nimport signal\nimport socket\nimport sys\n\ntry:\n    from cStringIO import StringIO  # py2\nexcept ImportError:\n    from io import StringIO  # py3\n\n# Tornado's own test suite requires the updated unittest module\n# (either py27+ or unittest2) so tornado.test.util enforces\n# this requirement, but for other users of tornado.testing we want\n# to allow the older version if unitest2 is not available.\ntry:\n    import unittest2 as unittest\nexcept ImportError:\n    import unittest\n\n_next_port = 10000\n\n\ndef get_unused_port():\n    \"\"\"Returns a (hopefully) unused port number.\n\n    This function does not guarantee that the port it returns is available,\n    only that a series of get_unused_port calls in a single process return\n    distinct ports.\n\n    **Deprecated**.  Use bind_unused_port instead, which is guaranteed\n    to find an unused port.\n    \"\"\"\n    global _next_port\n    port = _next_port\n    _next_port = _next_port + 1\n    return port\n\n\ndef bind_unused_port():\n    \"\"\"Binds a server socket to an available port on localhost.\n\n    Returns a tuple (socket, port).\n    \"\"\"\n    [sock] = netutil.bind_sockets(0, 'localhost', family=socket.AF_INET)\n    port = sock.getsockname()[1]\n    return sock, port\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 class provides\n    the `stop()` and `wait()` methods for this purpose.  The test\n    method itself must call ``self.wait()``, and asynchronous\n    callbacks should call ``self.stop()`` to signal completion.\n    Alternately, the `gen_test` decorator can be used to use yield points\n    from the `tornado.gen` module.\n\n    By default, a new `.IOLoop` is constructed for each test and is available\n    as ``self.io_loop``.  This `.IOLoop` should be used in the construction of\n    HTTP clients/servers, etc.  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 argument passing between self.stop and self.wait.\n        class MyTestCase(AsyncTestCase):\n            def test_http_fetch(self):\n                client = AsyncHTTPClient(self.io_loop)\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        # This test uses an explicit callback-based style.\n        class MyTestCase2(AsyncTestCase):\n            def test_http_fetch(self):\n                client = AsyncHTTPClient(self.io_loop)\n                client.fetch(\"http://www.tornadoweb.org/\", self.handle_fetch)\n                self.wait()\n\n            def handle_fetch(self, response):\n                # Test contents of response (failures and exceptions here\n                # will cause self.wait() to throw an exception and end the\n                # test).\n                # Exceptions thrown here are magically propagated to\n                # self.wait() in test_http_fetch() via stack_context.\n                self.assertIn(\"FriendFeed\", response.body)\n                self.stop()\n    \"\"\"\n    def __init__(self, *args, **kwargs):\n        super(AsyncTestCase, self).__init__(*args, **kwargs)\n        self.__stopped = False\n        self.__running = False\n        self.__failure = None\n        self.__stop_args = None\n        self.__timeout = None\n\n    def setUp(self):\n        super(AsyncTestCase, self).setUp()\n        self.io_loop = self.get_new_ioloop()\n        self.io_loop.make_current()\n\n    def tearDown(self):\n        self.io_loop.clear_current()\n        if (not IOLoop.initialized() or\n                self.io_loop is not IOLoop.instance()):\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(AsyncTestCase, self).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):\n        \"\"\"Creates a new `.IOLoop` for this test.  May be overridden in\n        subclasses for tests that require a specific `.IOLoop` (usually\n        the singleton `.IOLoop.instance()`).\n        \"\"\"\n        return IOLoop()\n\n    def _handle_exception(self, typ, value, tb):\n        self.__failure = sys.exc_info()\n        self.stop()\n        return True\n\n    def __rethrow(self):\n        if self.__failure is not None:\n            failure = self.__failure\n            self.__failure = None\n            raise_exc_info(failure)\n\n    def run(self, result=None):\n        with ExceptionStackContext(self._handle_exception):\n            super(AsyncTestCase, self).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\n    def stop(self, _arg=None, **kwargs):\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        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(self, condition=None, timeout=5):\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.\n\n        If ``condition`` is not None, the `.IOLoop` will be restarted\n        after `stop()` until ``condition()`` returns true.\n        \"\"\"\n        if not self.__stopped:\n            if timeout:\n                def timeout_func():\n                    try:\n                        raise self.failureException(\n                            'Async operation timed out after %s seconds' %\n                            timeout)\n                    except Exception:\n                        self.__failure = sys.exc_info()\n                    self.stop()\n                self.__timeout = self.io_loop.add_timeout(self.io_loop.time() + timeout, timeout_func)\n            while True:\n                self.__running = True\n                self.io_loop.start()\n                if (self.__failure is not None or\n                        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::\n\n        class MyHTTPTest(AsyncHTTPTestCase):\n            def get_app(self):\n                return Application([('/', MyHandler)...])\n\n            def test_homepage(self):\n                # The following two lines are equivalent to\n                #   response = self.fetch('/')\n                # but are shown in full here to demonstrate explicit use\n                # of self.stop and self.wait.\n                self.http_client.fetch(self.get_url('/'), self.stop)\n                response = self.wait()\n                # test contents of response\n    \"\"\"\n    def setUp(self):\n        super(AsyncHTTPTestCase, self).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):\n        return AsyncHTTPClient(io_loop=self.io_loop)\n\n    def get_http_server(self):\n        return HTTPServer(self._app, io_loop=self.io_loop,\n                          **self.get_httpserver_options())\n\n    def get_app(self):\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(self, path, **kwargs):\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 kwargs will be passed directly to\n        `.AsyncHTTPClient.fetch` (and so could be used to pass\n        ``method=\"POST\"``, ``body=\"...\"``, etc).\n        \"\"\"\n        self.http_client.fetch(self.get_url(path), self.stop, **kwargs)\n        return self.wait()\n\n    def get_httpserver_options(self):\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):\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):\n        return 'http'\n\n    def get_url(self, path):\n        \"\"\"Returns an absolute url for the given path on the test server.\"\"\"\n        return '%s://localhost:%s%s' % (self.get_protocol(),\n                                        self.get_http_port(), path)\n\n    def tearDown(self):\n        self.http_server.stop()\n        if (not IOLoop.initialized() or\n                self.http_client.io_loop is not IOLoop.instance()):\n            self.http_client.close()\n        super(AsyncHTTPTestCase, self).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    def get_http_client(self):\n        # Some versions of libcurl have deadlock bugs with ssl,\n        # so always run these tests with SimpleAsyncHTTPClient.\n        return SimpleAsyncHTTPClient(io_loop=self.io_loop, force_instance=True,\n                                     defaults=dict(validate_cert=False))\n\n    def get_httpserver_options(self):\n        return dict(ssl_options=self.get_ssl_options())\n\n    def get_ssl_options(self):\n        \"\"\"May be overridden by subclasses to select SSL options.\n\n        By default includes a self-signed testing certificate.\n        \"\"\"\n        # Testing keys were generated with:\n        # openssl req -new -keyout tornado/test/test.key -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    def get_protocol(self):\n        return 'https'\n\n\ndef gen_test(f):\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 gen.Task(self.fetch('/'))\n\n    \"\"\"\n    f = gen.coroutine(f)\n\n    @functools.wraps(f)\n    def wrapper(self):\n        return self.io_loop.run_sync(functools.partial(f, self), timeout=5)\n    return wrapper\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\n\n\nclass LogTrapTestCase(unittest.TestCase):\n    \"\"\"A test case that captures and discards all logging output\n    if the test passes.\n\n    Some libraries can produce a lot of logging output even when\n    the test succeeds, so this class can be useful to minimize the noise.\n    Simply use it as a base class for your test case.  It is safe to combine\n    with AsyncTestCase via multiple inheritance\n    (``class MyTestCase(AsyncHTTPTestCase, LogTrapTestCase):``)\n\n    This class assumes that only one log handler is configured and\n    that it is a `~logging.StreamHandler`.  This is true for both\n    `logging.basicConfig` and the \"pretty logging\" configured by\n    `tornado.options`.  It is not compatible with other log buffering\n    mechanisms, such as those provided by some test runners.\n    \"\"\"\n    def run(self, result=None):\n        logger = logging.getLogger()\n        if not logger.handlers:\n            logging.basicConfig()\n        handler = logger.handlers[0]\n        if (len(logger.handlers) > 1 or\n                not isinstance(handler, logging.StreamHandler)):\n            # Logging has been configured in a way we don't recognize,\n            # so just leave it alone.\n            super(LogTrapTestCase, self).run(result)\n            return\n        old_stream = handler.stream\n        try:\n            handler.stream = StringIO()\n            gen_log.info(\"RUNNING TEST: \" + str(self))\n            old_error_count = len(result.failures) + len(result.errors)\n            super(LogTrapTestCase, self).run(result)\n            new_error_count = len(result.failures) + len(result.errors)\n            if new_error_count != old_error_count:\n                old_stream.write(handler.stream.getvalue())\n        finally:\n            handler.stream = old_stream\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    Usage::\n\n        with ExpectLog('tornado.application', \"Uncaught exception\"):\n            error_response = self.fetch(\"/some_page\")\n    \"\"\"\n    def __init__(self, logger, regex, required=True):\n        \"\"\"Constructs an ExpectLog context manager.\n\n        :param logger: Logger object (or name of logger) to watch.  Pass\n            an empty string to watch the root logger.\n        :param regex: Regular expression to match.  Any log entries on\n            the specified logger that match this regex will be suppressed.\n        :param required: If true, an exeption will be raised if the end of\n            the ``with`` statement is reached without matching any log entries.\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        self.matched = False\n\n    def filter(self, record):\n        message = record.getMessage()\n        if self.regex.match(message):\n            self.matched = True\n            return False\n        return True\n\n    def __enter__(self):\n        self.logger.addFilter(self)\n\n    def __exit__(self, typ, value, tb):\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\n\ndef main(**kwargs):\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.\n\n    The easiest way to run a test is via the command line::\n\n        python -m tornado.testing tornado.test.stack_context_test\n\n    See the standard library unittest module for ways in which tests can\n    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.stack_context_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    from tornado.options import define, options, parse_command_line\n\n    define('exception_on_interrupt', type=bool, default=True,\n           help=(\"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    # 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    try:\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)\n        else:\n            unittest.main(defaultTest=\"all\", argv=argv, **kwargs)\n    except SystemExit as e:\n        if e.code == 0:\n            gen_log.info('PASS')\n        else:\n            gen_log.error('FAIL')\n        raise\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:** `gen_test` Decorator Does Not Accept `timeout` Parameter, Causing TypeError

**Description:**
When attempting to use the `gen_test` decorator with the `timeout` parameter to specify a custom timeout for asynchronous tests, a `TypeError` is raised. This prevents tests that require longer execution times from running successfully.

**Example Code:**
```python
from tornado import gen, ioloop
from tornado.testing import AsyncTestCase, gen_test

class MyAsyncTests(AsyncTestCase):

    @gen_test(timeout=10)
    def test_long_operation(self):
        # Simulate a long-running asynchronous operation
        yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 15)
```

**Expected Behavior:**
The `timeout` parameter should allow configuring the maximum duration the test can run before timing out. In this example, the test should allow up to 10 seconds for `test_long_operation` to complete.

**Actual Behavior:**
Running the test results in the following error:
```
TypeError: gen_test() got an unexpected keyword argument 'timeout'
```
This error indicates that the `gen_test` decorator does not recognize the `timeout` parameter, causing the test to fail immediately instead of adhering to the specified timeout.

**[/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
#!/usr/bin/env python
"""Support classes for automated testing.

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

* `ExpectLog` and `LogTrapTestCase`: 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.
"""

from __future__ import absolute_import, division, print_function, with_statement

try:
    from tornado import gen
    from tornado.httpclient import AsyncHTTPClient
    from tornado.httpserver import HTTPServer
    from tornado.simple_httpclient import SimpleAsyncHTTPClient
    from tornado.ioloop import IOLoop
    from tornado import netutil
except ImportError:
    # These modules are not importable on app engine.  Parts of this module
    # won't work, but e.g. LogTrapTestCase and main() will.
    AsyncHTTPClient = None
    gen = None
    HTTPServer = None
    IOLoop = None
    netutil = None
    SimpleAsyncHTTPClient = None
from tornado.log import gen_log
from tornado.stack_context import ExceptionStackContext
from tornado.util import raise_exc_info, basestring_type
import functools
import logging
import os
import re
import signal
import socket
import sys

try:
    from cStringIO import StringIO  # py2
except ImportError:
    from io import StringIO  # py3

# Tornado's own test suite requires the updated unittest module
# (either py27+ or unittest2) so tornado.test.util enforces
# this requirement, but for other users of tornado.testing we want
# to allow the older version if unitest2 is not available.
try:
    import unittest2 as unittest
except ImportError:
    import unittest

_next_port = 10000


def get_unused_port():
    """Returns a (hopefully) unused port number.

    This function does not guarantee that the port it returns is available,
    only that a series of get_unused_port calls in a single process return
    distinct ports.

    **Deprecated**.  Use bind_unused_port instead, which is guaranteed
    to find an unused port.
    """
    global _next_port
    port = _next_port
    _next_port = _next_port + 1
    return port


def bind_unused_port():
    """Binds a server socket to an available port on localhost.

    Returns a tuple (socket, port).
    """
    [sock] = netutil.bind_sockets(0, 'localhost', family=socket.AF_INET)
    port = sock.getsockname()[1]
    return sock, port


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 class provides
    the `stop()` and `wait()` methods for this purpose.  The test
    method itself must call ``self.wait()``, and asynchronous
    callbacks should call ``self.stop()`` to signal completion.
    Alternately, the `gen_test` decorator can be used to use yield points
    from the `tornado.gen` module.

    By default, a new `.IOLoop` is constructed for each test and is available
    as ``self.io_loop``.  This `.IOLoop` should be used in the construction of
    HTTP clients/servers, etc.  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 argument passing between self.stop and self.wait.
        class MyTestCase(AsyncTestCase):
            def test_http_fetch(self):
                client = AsyncHTTPClient(self.io_loop)
                client.fetch("http://www.tornadoweb.org/", self.stop)
                response = self.wait()
                # Test contents of response
                self.assertIn("FriendFeed", response.body)

        # This test uses an explicit callback-based style.
        class MyTestCase2(AsyncTestCase):
            def test_http_fetch(self):
                client = AsyncHTTPClient(self.io_loop)
                client.fetch("http://www.tornadoweb.org/", self.handle_fetch)
                self.wait()

            def handle_fetch(self, response):
                # Test contents of response (failures and exceptions here
                # will cause self.wait() to throw an exception and end the
                # test).
                # Exceptions thrown here are magically propagated to
                # self.wait() in test_http_fetch() via stack_context.
                self.assertIn("FriendFeed", response.body)
                self.stop()
    """
    def __init__(self, *args, **kwargs):
        super(AsyncTestCase, self).__init__(*args, **kwargs)
        self.__stopped = False
        self.__running = False
        self.__failure = None
        self.__stop_args = None
        self.__timeout = None

    def setUp(self):
        super(AsyncTestCase, self).setUp()
        self.io_loop = self.get_new_ioloop()
        self.io_loop.make_current()

    def tearDown(self):
        self.io_loop.clear_current()
        if (not IOLoop.initialized() or
                self.io_loop is not IOLoop.instance()):
            # 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)
        super(AsyncTestCase, self).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):
        """Creates a new `.IOLoop` for this test.  May be overridden in
        subclasses for tests that require a specific `.IOLoop` (usually
        the singleton `.IOLoop.instance()`).
        """
        return IOLoop()

    def _handle_exception(self, typ, value, tb):
        self.__failure = sys.exc_info()
        self.stop()
        return True

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

    def run(self, result=None):
        with ExceptionStackContext(self._handle_exception):
            super(AsyncTestCase, self).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()

    def stop(self, _arg=None, **kwargs):
        """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()`.
        """
        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=None, timeout=5):
        """Runs the `.IOLoop` until stop is called or timeout has passed.

        In the event of a timeout, an exception will be thrown.

        If ``condition`` is not None, the `.IOLoop` will be restarted
        after `stop()` until ``condition()`` returns true.
        """
        if not self.__stopped:
            if timeout:
                def timeout_func():
                    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::

        class MyHTTPTest(AsyncHTTPTestCase):
            def get_app(self):
                return Application([('/', MyHandler)...])

            def test_homepage(self):
                # The following two lines are equivalent to
                #   response = self.fetch('/')
                # but are shown in full here to demonstrate explicit use
                # of self.stop and self.wait.
                self.http_client.fetch(self.get_url('/'), self.stop)
                response = self.wait()
                # test contents of response
    """
    def setUp(self):
        super(AsyncHTTPTestCase, self).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):
        return AsyncHTTPClient(io_loop=self.io_loop)

    def get_http_server(self):
        return HTTPServer(self._app, io_loop=self.io_loop,
                          **self.get_httpserver_options())

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

    def fetch(self, path, **kwargs):
        """Convenience method to synchronously fetch a url.

        The given path will be appended to the local server's host and
        port.  Any additional kwargs will be passed directly to
        `.AsyncHTTPClient.fetch` (and so could be used to pass
        ``method="POST"``, ``body="..."``, etc).
        """
        self.http_client.fetch(self.get_url(path), self.stop, **kwargs)
        return self.wait()

    def get_httpserver_options(self):
        """May be overridden by subclasses to return additional
        keyword arguments for the server.
        """
        return {}

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

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

    def get_protocol(self):
        return 'http'

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

    def tearDown(self):
        self.http_server.stop()
        if (not IOLoop.initialized() or
                self.http_client.io_loop is not IOLoop.instance()):
            self.http_client.close()
        super(AsyncHTTPTestCase, self).tearDown()


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

    Interface is generally the same as `AsyncHTTPTestCase`.
    """
    def get_http_client(self):
        # Some versions of libcurl have deadlock bugs with ssl,
        # so always run these tests with SimpleAsyncHTTPClient.
        return SimpleAsyncHTTPClient(io_loop=self.io_loop, force_instance=True,
                                     defaults=dict(validate_cert=False))

    def get_httpserver_options(self):
        return dict(ssl_options=self.get_ssl_options())

    def get_ssl_options(self):
        """May be overridden by subclasses to select SSL options.

        By default includes a self-signed testing certificate.
        """
        # 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):
        return 'https'


def gen_test(f):
    """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 gen.Task(self.fetch('/'))

    """
    f = gen.coroutine(f)

    @functools.wraps(f)
    def wrapper(self):
        return self.io_loop.run_sync(functools.partial(f, self), timeout=5)
    return wrapper


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


class LogTrapTestCase(unittest.TestCase):
    """A test case that captures and discards all logging output
    if the test passes.

    Some libraries can produce a lot of logging output even when
    the test succeeds, so this class can be useful to minimize the noise.
    Simply use it as a base class for your test case.  It is safe to combine
    with AsyncTestCase via multiple inheritance
    (``class MyTestCase(AsyncHTTPTestCase, LogTrapTestCase):``)

    This class assumes that only one log handler is configured and
    that it is a `~logging.StreamHandler`.  This is true for both
    `logging.basicConfig` and the "pretty logging" configured by
    `tornado.options`.  It is not compatible with other log buffering
    mechanisms, such as those provided by some test runners.
    """
    def run(self, result=None):
        logger = logging.getLogger()
        if not logger.handlers:
            logging.basicConfig()
        handler = logger.handlers[0]
        if (len(logger.handlers) > 1 or
                not isinstance(handler, logging.StreamHandler)):
            # Logging has been configured in a way we don't recognize,
            # so just leave it alone.
            super(LogTrapTestCase, self).run(result)
            return
        old_stream = handler.stream
        try:
            handler.stream = StringIO()
            gen_log.info("RUNNING TEST: " + str(self))
            old_error_count = len(result.failures) + len(result.errors)
            super(LogTrapTestCase, self).run(result)
            new_error_count = len(result.failures) + len(result.errors)
            if new_error_count != old_error_count:
                old_stream.write(handler.stream.getvalue())
        finally:
            handler.stream = old_stream


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.*

    Usage::

        with ExpectLog('tornado.application', "Uncaught exception"):
            error_response = self.fetch("/some_page")
    """
    def __init__(self, logger, regex, required=True):
        """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 exeption will be raised if the end of
            the ``with`` statement is reached without matching any log entries.
        """
        if isinstance(logger, basestring_type):
            logger = logging.getLogger(logger)
        self.logger = logger
        self.regex = re.compile(regex)
        self.required = required
        self.matched = False

    def filter(self, record):
        message = record.getMessage()
        if self.regex.match(message):
            self.matched = True
            return False
        return True

    def __enter__(self):
        self.logger.addFilter(self)

    def __exit__(self, typ, value, tb):
        self.logger.removeFilter(self)
        if not typ and self.required and not self.matched:
            raise Exception("did not get expected log message")


def main(**kwargs):
    """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.

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

        python -m tornado.testing tornado.test.stack_context_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.stack_context_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.
    """
    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)
    try:
        # 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)
        else:
            unittest.main(defaultTest="all", argv=argv, **kwargs)
    except SystemExit as e:
        if e.code == 0:
            gen_log.info('PASS')
        else:
            gen_log.error('FAIL')
        raise

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