From 04f832bae166bb5482552156973a6e65b3866ef8 Mon Sep 17 00:00:00 2001 From: Ben Darnell Date: Sat, 22 Jul 2023 21:42:22 -0400 Subject: [PATCH] autoreload_test: Reduce repetition --- tornado/test/autoreload_test.py | 212 +++++++++++++++----------------- 1 file changed, 100 insertions(+), 112 deletions(-) diff --git a/tornado/test/autoreload_test.py b/tornado/test/autoreload_test.py index c08c5be36..ff934520d 100644 --- a/tornado/test/autoreload_test.py +++ b/tornado/test/autoreload_test.py @@ -4,6 +4,7 @@ import subprocess from subprocess import Popen import sys from tempfile import mkdtemp +import textwrap import time import unittest @@ -12,6 +13,32 @@ class AutoreloadTest(unittest.TestCase): def setUp(self): self.path = mkdtemp() + # Each test app runs itself twice via autoreload. The first time it manually triggers + # a reload (could also do this by touching a file but this is faster since filesystem + # timestamps are not necessarily high resolution). The second time it exits directly + # so that the autoreload wrapper (if it is used) doesn't catch it. + # + # The last line of each test's "main" program should be + # exec(open("run_twice_magic.py").read()) + self.write_files( + { + "run_twice_magic.py": """ + import os + import sys + + import tornado.autoreload + + sys.stdout.flush() + + if "TESTAPP_STARTED" not in os.environ: + os.environ["TESTAPP_STARTED"] = "1" + tornado.autoreload._reload() + else: + os._exit(0) + """ + } + ) + def tearDown(self): try: shutil.rmtree(self.path) @@ -24,13 +51,56 @@ class AutoreloadTest(unittest.TestCase): time.sleep(1) shutil.rmtree(self.path) + def write_files(self, tree, base_path=None): + """Write a directory tree to self.path. + + tree is a dictionary mapping file names to contents, or + sub-dictionaries representing subdirectories. + """ + if base_path is None: + base_path = self.path + for name, contents in tree.items(): + if isinstance(contents, dict): + os.mkdir(os.path.join(base_path, name)) + self.write_files(contents, os.path.join(base_path, name)) + else: + with open(os.path.join(base_path, name), "w", encoding="utf-8") as f: + f.write(textwrap.dedent(contents)) + + def run_subprocess(self, args): + # Make sure the tornado module under test is available to the test + # application + pythonpath = os.getcwd() + if "PYTHONPATH" in os.environ: + pythonpath += os.pathsep + os.environ["PYTHONPATH"] + + p = Popen( + args, + stdout=subprocess.PIPE, + env=dict(os.environ, PYTHONPATH=pythonpath), + cwd=self.path, + universal_newlines=True, + encoding="utf-8", + ) + + # This timeout needs to be fairly generous for pypy due to jit + # warmup costs. + for i in range(40): + if p.poll() is not None: + break + time.sleep(0.1) + else: + p.kill() + raise Exception("subprocess failed to terminate") + + out = p.communicate()[0] + self.assertEqual(p.returncode, 0) + return out + def test_reload(self): main = """\ -import os import sys -from tornado import autoreload - # In module mode, the path is set to the parent directory and we can import testapp. try: import testapp @@ -41,40 +111,23 @@ else: spec = getattr(sys.modules[__name__], '__spec__', None) print(f"Starting {__name__=}, __spec__.name={getattr(spec, 'name', None)}") -if 'TESTAPP_STARTED' not in os.environ: - os.environ['TESTAPP_STARTED'] = '1' - sys.stdout.flush() - autoreload._reload() +exec(open("run_twice_magic.py").read()) """ # Create temporary test application - os.mkdir(os.path.join(self.path, "testapp")) - open( - os.path.join(self.path, "testapp/__init__.py"), "w", encoding="utf-8" - ).close() - with open( - os.path.join(self.path, "testapp/__main__.py"), "w", encoding="utf-8" - ) as f: - f.write(main) - - # Make sure the tornado module under test is available to the test - # application - pythonpath = os.getcwd() - if "PYTHONPATH" in os.environ: - pythonpath += os.pathsep + os.environ["PYTHONPATH"] + self.write_files( + { + "testapp": { + "__init__.py": "", + "__main__.py": main, + }, + } + ) with self.subTest(mode="module"): # In module mode, the path is set to the parent directory and we can import testapp. # Also, the __spec__.name is set to the fully qualified module name. - p = Popen( - [sys.executable, "-m", "testapp"], - stdout=subprocess.PIPE, - cwd=self.path, - env=dict(os.environ, PYTHONPATH=pythonpath), - universal_newlines=True, - encoding="utf-8", - ) - out = p.communicate()[0] + out = self.run_subprocess([sys.executable, "-m", "testapp"]) self.assertEqual( out, ( @@ -87,15 +140,7 @@ if 'TESTAPP_STARTED' not in os.environ: with self.subTest(mode="file"): # When the __main__.py file is run directly, there is no qualified module spec and we # cannot import testapp. - p = Popen( - [sys.executable, "testapp/__main__.py"], - stdout=subprocess.PIPE, - cwd=self.path, - env=dict(os.environ, PYTHONPATH=pythonpath), - universal_newlines=True, - encoding="utf-8", - ) - out = p.communicate()[0] + out = self.run_subprocess([sys.executable, "testapp/__main__.py"]) self.assertEqual( out, "import testapp failed\nStarting __name__='__main__', __spec__.name=None\n" @@ -105,15 +150,7 @@ if 'TESTAPP_STARTED' not in os.environ: with self.subTest(mode="directory"): # Running as a directory finds __main__.py like a module. It does not manipulate # sys.path but it does set a spec with a name of exactly __main__. - p = Popen( - [sys.executable, "testapp"], - stdout=subprocess.PIPE, - cwd=self.path, - env=dict(os.environ, PYTHONPATH=pythonpath), - universal_newlines=True, - encoding="utf-8", - ) - out = p.communicate()[0] + out = self.run_subprocess([sys.executable, "testapp"]) self.assertEqual( out, "import testapp failed\nStarting __name__='__main__', __spec__.name=__main__\n" @@ -125,7 +162,6 @@ if 'TESTAPP_STARTED' not in os.environ: # is used on an application that also has an internal # autoreload, the reload wrapper is preserved on restart. main = """\ -import os import sys # This import will fail if path is not set up correctly @@ -134,78 +170,38 @@ import testapp if 'tornado.autoreload' not in sys.modules: raise Exception('started without autoreload wrapper') -import tornado.autoreload - print('Starting') -sys.stdout.flush() -if 'TESTAPP_STARTED' not in os.environ: - os.environ['TESTAPP_STARTED'] = '1' - # Simulate an internal autoreload (one not caused - # by the wrapper). - tornado.autoreload._reload() -else: - # Exit directly so autoreload doesn't catch it. - os._exit(0) +exec(open("run_twice_magic.py").read()) """ - # Create temporary test application - os.mkdir(os.path.join(self.path, "testapp")) - init_file = os.path.join(self.path, "testapp", "__init__.py") - open(init_file, "w", encoding="utf-8").close() - main_file = os.path.join(self.path, "testapp", "__main__.py") - with open(main_file, "w", encoding="utf-8") as f: - f.write(main) - - # Make sure the tornado module under test is available to the test - # application - pythonpath = os.getcwd() - if "PYTHONPATH" in os.environ: - pythonpath += os.pathsep + os.environ["PYTHONPATH"] - - autoreload_proc = Popen( - [sys.executable, "-m", "tornado.autoreload", "-m", "testapp"], - stdout=subprocess.PIPE, - cwd=self.path, - env=dict(os.environ, PYTHONPATH=pythonpath), - universal_newlines=True, - encoding="utf-8", + self.write_files( + { + "testapp": { + "__init__.py": "", + "__main__.py": main, + }, + } ) - # This timeout needs to be fairly generous for pypy due to jit - # warmup costs. - for i in range(40): - if autoreload_proc.poll() is not None: - break - time.sleep(0.1) - else: - autoreload_proc.kill() - raise Exception("subprocess failed to terminate") - - out = autoreload_proc.communicate()[0] + out = self.run_subprocess( + [sys.executable, "-m", "tornado.autoreload", "-m", "testapp"] + ) self.assertEqual(out, "Starting\n" * 2) def test_reload_wrapper_args(self): main = """\ -import os import sys print(os.path.basename(sys.argv[0])) print(f'argv={sys.argv[1:]}') -sys.stdout.flush() -os._exit(0) +exec(open("run_twice_magic.py").read()) """ # Create temporary test application - main_file = os.path.join(self.path, "main.py") - with open(main_file, "w", encoding="utf-8") as f: - f.write(main) + self.write_files({"main.py": main}) # Make sure the tornado module under test is available to the test # application - pythonpath = os.getcwd() - if "PYTHONPATH" in os.environ: - pythonpath += os.pathsep + os.environ["PYTHONPATH"] - - autoreload_proc = Popen( + out = self.run_subprocess( [ sys.executable, "-m", @@ -216,14 +212,6 @@ os._exit(0) "-m", "arg3", ], - stdout=subprocess.PIPE, - cwd=self.path, - env=dict(os.environ, PYTHONPATH=pythonpath), - universal_newlines=True, - encoding="utf-8", ) - out, err = autoreload_proc.communicate() - if err: - print("subprocess stderr:", err) - self.assertEqual(out, "main.py\nargv=['arg1', '--arg2', '-m', 'arg3']\n") + self.assertEqual(out, "main.py\nargv=['arg1', '--arg2', '-m', 'arg3']\n" * 2) -- 2.47.2