Antoine Pitrou | cb4f929 | 2010-11-10 14:01:16 +0000 | [diff] [blame^] | 1 | # Common utility functions used by various script execution tests |
| 2 | # e.g. test_cmd_line, test_cmd_line_script and test_runpy |
| 3 | |
| 4 | import sys |
| 5 | import os |
| 6 | import os.path |
| 7 | import tempfile |
| 8 | import subprocess |
| 9 | import py_compile |
| 10 | import contextlib |
| 11 | import shutil |
| 12 | import zipfile |
| 13 | |
| 14 | |
| 15 | # Executing the interpreter in a subprocess |
| 16 | def _assert_python(expected_success, *args, **env_vars): |
| 17 | cmd_line = [sys.executable] |
| 18 | if not env_vars: |
| 19 | cmd_line.append('-E') |
| 20 | cmd_line.extend(args) |
| 21 | # Need to preserve the original environment, for in-place testing of |
| 22 | # shared library builds. |
| 23 | env = os.environ.copy() |
| 24 | env.update(env_vars) |
| 25 | p = subprocess.Popen(cmd_line, stdin=subprocess.PIPE, |
| 26 | stdout=subprocess.PIPE, stderr=subprocess.PIPE, |
| 27 | env=env) |
| 28 | try: |
| 29 | out, err = p.communicate() |
| 30 | finally: |
| 31 | subprocess._cleanup() |
| 32 | p.stdout.close() |
| 33 | p.stderr.close() |
| 34 | rc = p.returncode |
| 35 | if (rc and expected_success) or (not rc and not expected_success): |
| 36 | raise AssertionError( |
| 37 | "Process return code is %d, " |
| 38 | "stderr follows:\n%s" % (rc, err.decode('ascii', 'ignore'))) |
| 39 | return rc, out, err |
| 40 | |
| 41 | def assert_python_ok(*args, **env_vars): |
| 42 | """ |
| 43 | Assert that running the interpreter with `args` and optional environment |
| 44 | variables `env_vars` is ok and return a (return code, stdout, stderr) tuple. |
| 45 | """ |
| 46 | return _assert_python(True, *args, **env_vars) |
| 47 | |
| 48 | def assert_python_failure(*args, **env_vars): |
| 49 | """ |
| 50 | Assert that running the interpreter with `args` and optional environment |
| 51 | variables `env_vars` fails and return a (return code, stdout, stderr) tuple. |
| 52 | """ |
| 53 | return _assert_python(False, *args, **env_vars) |
| 54 | |
| 55 | def spawn_python(*args): |
| 56 | cmd_line = [sys.executable, '-E'] |
| 57 | cmd_line.extend(args) |
| 58 | return subprocess.Popen(cmd_line, stdin=subprocess.PIPE, |
| 59 | stdout=subprocess.PIPE, stderr=subprocess.STDOUT) |
| 60 | |
| 61 | def kill_python(p): |
| 62 | p.stdin.close() |
| 63 | data = p.stdout.read() |
| 64 | p.stdout.close() |
| 65 | # try to cleanup the child so we don't appear to leak when running |
| 66 | # with regrtest -R. |
| 67 | p.wait() |
| 68 | subprocess._cleanup() |
| 69 | return data |
| 70 | |
| 71 | # Script creation utilities |
| 72 | @contextlib.contextmanager |
| 73 | def temp_dir(): |
| 74 | dirname = tempfile.mkdtemp() |
| 75 | dirname = os.path.realpath(dirname) |
| 76 | try: |
| 77 | yield dirname |
| 78 | finally: |
| 79 | shutil.rmtree(dirname) |
| 80 | |
| 81 | def make_script(script_dir, script_basename, source): |
| 82 | script_filename = script_basename+os.extsep+'py' |
| 83 | script_name = os.path.join(script_dir, script_filename) |
| 84 | # The script should be encoded to UTF-8, the default string encoding |
| 85 | script_file = open(script_name, 'w', encoding='utf-8') |
| 86 | script_file.write(source) |
| 87 | script_file.close() |
| 88 | return script_name |
| 89 | |
| 90 | def make_pkg(pkg_dir, init_source=''): |
| 91 | os.mkdir(pkg_dir) |
| 92 | make_script(pkg_dir, '__init__', init_source) |
| 93 | |
| 94 | def make_zip_pkg(zip_dir, zip_basename, pkg_name, script_basename, |
| 95 | source, depth=1, compiled=False): |
| 96 | unlink = [] |
| 97 | init_name = make_script(zip_dir, '__init__', '') |
| 98 | unlink.append(init_name) |
| 99 | init_basename = os.path.basename(init_name) |
| 100 | script_name = make_script(zip_dir, script_basename, source) |
| 101 | unlink.append(script_name) |
| 102 | if compiled: |
| 103 | init_name = py_compile(init_name, doraise=True) |
| 104 | script_name = py_compile(script_name, doraise=True) |
| 105 | unlink.extend((init_name, script_name)) |
| 106 | pkg_names = [os.sep.join([pkg_name]*i) for i in range(1, depth+1)] |
| 107 | script_name_in_zip = os.path.join(pkg_names[-1], os.path.basename(script_name)) |
| 108 | zip_filename = zip_basename+os.extsep+'zip' |
| 109 | zip_name = os.path.join(zip_dir, zip_filename) |
| 110 | zip_file = zipfile.ZipFile(zip_name, 'w') |
| 111 | for name in pkg_names: |
| 112 | init_name_in_zip = os.path.join(name, init_basename) |
| 113 | zip_file.write(init_name, init_name_in_zip) |
| 114 | zip_file.write(script_name, script_name_in_zip) |
| 115 | zip_file.close() |
| 116 | for name in unlink: |
| 117 | os.unlink(name) |
| 118 | #if test.support.verbose: |
| 119 | # zip_file = zipfile.ZipFile(zip_name, 'r') |
| 120 | # print 'Contents of %r:' % zip_name |
| 121 | # zip_file.printdir() |
| 122 | # zip_file.close() |
| 123 | return zip_name, os.path.join(zip_name, script_name_in_zip) |