from subprocess import Popen
import sys
from tempfile import mkdtemp
+import textwrap
import time
import unittest
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)
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
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,
(
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"
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"
# 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
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",
"-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)
projects / thirdparty / tornado.git / commitdiff
