codekingpro/portable-devtools

1import os
2import shutil
3import subprocess
4from subprocess import Popen
5import sys
6from tempfile import mkdtemp
7import textwrap
8import time
9import unittest
10
11
12class AutoreloadTest(unittest.TestCase):
13    def setUp(self):
14        # When these tests fail the output sometimes exceeds the default maxDiff.
15        self.maxDiff = 1024
16
17        self.path = mkdtemp()
18
19        # Most test apps run themselves twice via autoreload. The first time it manually triggers
20        # a reload (could also do this by touching a file but this is faster since filesystem
21        # timestamps are not necessarily high resolution). The second time it exits directly
22        # so that the autoreload wrapper (if it is used) doesn't catch it.
23        #
24        # The last line of each such test's "main" program should be
25        #     exec(open("run_twice_magic.py").read())
26        self.write_files(
27            {
28                "run_twice_magic.py": """
29                    import os
30                    import sys
31
32                    import tornado.autoreload
33
34                    sys.stdout.flush()
35
36                    if "TESTAPP_STARTED" not in os.environ:
37                        os.environ["TESTAPP_STARTED"] = "1"
38                        tornado.autoreload._reload()
39                    else:
40                        os._exit(0)
41                """
42            }
43        )
44
45    def tearDown(self):
46        try:
47            shutil.rmtree(self.path)
48        except OSError:
49            # Windows disallows deleting files that are in use by
50            # another process, and even though we've waited for our
51            # child process below, it appears that its lock on these
52            # files is not guaranteed to be released by this point.
53            # Sleep and try again (once).
54            time.sleep(1)
55            shutil.rmtree(self.path)
56
57    def write_files(self, tree, base_path=None):
58        """Write a directory tree to self.path.
59
60        tree is a dictionary mapping file names to contents, or
61        sub-dictionaries representing subdirectories.
62        """
63        if base_path is None:
64            base_path = self.path
65        for name, contents in tree.items():
66            if isinstance(contents, dict):
67                os.mkdir(os.path.join(base_path, name))
68                self.write_files(contents, os.path.join(base_path, name))
69            else:
70                with open(os.path.join(base_path, name), "w", encoding="utf-8") as f:
71                    f.write(textwrap.dedent(contents))
72
73    def run_subprocess(self, args):
74        # Make sure the tornado module under test is available to the test
75        # application
76        parts = [os.getcwd()]
77        if "PYTHONPATH" in os.environ:
78            parts += [
79                os.path.join(os.getcwd(), part)
80                for part in os.environ["PYTHONPATH"].split(os.pathsep)
81            ]
82        pythonpath = os.pathsep.join(parts)
83
84        p = Popen(
85            args,
86            stdout=subprocess.PIPE,
87            env=dict(os.environ, PYTHONPATH=pythonpath),
88            cwd=self.path,
89            universal_newlines=True,
90            encoding="utf-8",
91        )
92
93        # This timeout needs to be fairly generous for pypy due to jit
94        # warmup costs.
95        for i in range(40):
96            if p.poll() is not None:
97                break
98            time.sleep(0.1)
99        else:
100            p.kill()
101            raise Exception("subprocess failed to terminate")
102
103        out = p.communicate()[0]
104        self.assertEqual(p.returncode, 0)
105        return out
106
107    def test_reload(self):
108        main = """\
109import sys
110
111# In module mode, the path is set to the parent directory and we can import testapp.
112try:
113    import testapp
114except ImportError:
115    print("import testapp failed")
116else:
117    print("import testapp succeeded")
118
119spec = getattr(sys.modules[__name__], '__spec__', None)
120print(f"Starting {__name__=}, __spec__.name={getattr(spec, 'name', None)}")
121exec(open("run_twice_magic.py", encoding="utf-8").read())
122"""
123
124        # Create temporary test application
125        self.write_files(
126            {
127                "testapp": {
128                    "__init__.py": "",
129                    "__main__.py": main,
130                },
131            }
132        )
133
134        # The autoreload wrapper should support all the same modes as the python interpreter.
135        # The wrapper itself should have no effect on this test so we try all modes with and
136        # without it.
137        for wrapper in [False, True]:
138            with self.subTest(wrapper=wrapper):
139                with self.subTest(mode="module"):
140                    if wrapper:
141                        base_args = [sys.executable, "-m", "tornado.autoreload"]
142                    else:
143                        base_args = [sys.executable]
144                    # In module mode, the path is set to the parent directory and we can import
145                    # testapp. Also, the __spec__.name is set to the fully qualified module name.
146                    out = self.run_subprocess(base_args + ["-m", "testapp"])
147                    self.assertEqual(
148                        out,
149                        (
150                            "import testapp succeeded\n"
151                            + "Starting __name__='__main__', __spec__.name=testapp.__main__\n"
152                        )
153                        * 2,
154                    )
155
156                with self.subTest(mode="file"):
157                    out = self.run_subprocess(base_args + ["testapp/__main__.py"])
158                    # In file mode, we do not expect the path to be set so we can import testapp,
159                    # but when the wrapper is used the -m argument to the python interpreter
160                    # does this for us.
161                    expect_import = (
162                        "import testapp succeeded"
163                        if wrapper
164                        else "import testapp failed"
165                    )
166                    # In file mode there is no qualified module spec.
167                    self.assertEqual(
168                        out,
169                        f"{expect_import}\nStarting __name__='__main__', __spec__.name=None\n"
170                        * 2,
171                    )
172
173                with self.subTest(mode="directory"):
174                    # Running as a directory finds __main__.py like a module. It does not manipulate
175                    # sys.path but it does set a spec with a name of exactly __main__.
176                    out = self.run_subprocess(base_args + ["testapp"])
177                    expect_import = (
178                        "import testapp succeeded"
179                        if wrapper
180                        else "import testapp failed"
181                    )
182                    self.assertEqual(
183                        out,
184                        f"{expect_import}\nStarting __name__='__main__', __spec__.name=__main__\n"
185                        * 2,
186                    )
187
188    def test_reload_wrapper_preservation(self):
189        # This test verifies that when `python -m tornado.autoreload`
190        # is used on an application that also has an internal
191        # autoreload, the reload wrapper is preserved on restart.
192        main = """\
193import sys
194
195# This import will fail if path is not set up correctly
196import testapp
197
198if 'tornado.autoreload' not in sys.modules:
199    raise Exception('started without autoreload wrapper')
200
201print('Starting')
202exec(open("run_twice_magic.py", encoding="utf-8").read())
203"""
204
205        self.write_files(
206            {
207                "testapp": {
208                    "__init__.py": "",
209                    "__main__.py": main,
210                },
211            }
212        )
213
214        out = self.run_subprocess(
215            [sys.executable, "-m", "tornado.autoreload", "-m", "testapp"]
216        )
217        self.assertEqual(out, "Starting\n" * 2)
218
219    def test_reload_wrapper_args(self):
220        main = """\
221import os
222import sys
223
224print(os.path.basename(sys.argv[0]))
225print(f'argv={sys.argv[1:]}')
226exec(open("run_twice_magic.py", encoding="utf-8").read())
227"""
228        # Create temporary test application
229        self.write_files({"main.py": main})
230
231        # Make sure the tornado module under test is available to the test
232        # application
233        out = self.run_subprocess(
234            [
235                sys.executable,
236                "-m",
237                "tornado.autoreload",
238                "main.py",
239                "arg1",
240                "--arg2",
241                "-m",
242                "arg3",
243            ],
244        )
245
246        self.assertEqual(out, "main.py\nargv=['arg1', '--arg2', '-m', 'arg3']\n" * 2)
247
248    def test_reload_wrapper_until_success(self):
249        main = """\
250import os
251import sys
252
253if "TESTAPP_STARTED" in os.environ:
254    print("exiting cleanly")
255    sys.exit(0)
256else:
257    print("reloading")
258    exec(open("run_twice_magic.py", encoding="utf-8").read())
259"""
260
261        # Create temporary test application
262        self.write_files({"main.py": main})
263
264        out = self.run_subprocess(
265            [sys.executable, "-m", "tornado.autoreload", "--until-success", "main.py"]
266        )
267
268        self.assertEqual(out, "reloading\nexiting cleanly\n")
269