mirror of https://github.com/python/cpython
1365 lines
50 KiB
Python
1365 lines
50 KiB
Python
import gc
|
|
import json
|
|
import importlib
|
|
import importlib.util
|
|
import os
|
|
import os.path
|
|
import py_compile
|
|
import sys
|
|
from test import support
|
|
from test.support import import_helper
|
|
from test.support import os_helper
|
|
from test.support import script_helper
|
|
from test.support import warnings_helper
|
|
import textwrap
|
|
import types
|
|
import unittest
|
|
import warnings
|
|
imp = warnings_helper.import_deprecated('imp')
|
|
import _imp
|
|
import _testinternalcapi
|
|
try:
|
|
import _xxsubinterpreters as _interpreters
|
|
except ModuleNotFoundError:
|
|
_interpreters = None
|
|
|
|
|
|
OS_PATH_NAME = os.path.__name__
|
|
|
|
|
|
def requires_subinterpreters(meth):
|
|
"""Decorator to skip a test if subinterpreters are not supported."""
|
|
return unittest.skipIf(_interpreters is None,
|
|
'subinterpreters required')(meth)
|
|
|
|
|
|
def requires_load_dynamic(meth):
|
|
"""Decorator to skip a test if not running under CPython or lacking
|
|
imp.load_dynamic()."""
|
|
meth = support.cpython_only(meth)
|
|
return unittest.skipIf(getattr(imp, 'load_dynamic', None) is None,
|
|
'imp.load_dynamic() required')(meth)
|
|
|
|
|
|
class ModuleSnapshot(types.SimpleNamespace):
|
|
"""A representation of a module for testing.
|
|
|
|
Fields:
|
|
|
|
* id - the module's object ID
|
|
* module - the actual module or an adequate substitute
|
|
* __file__
|
|
* __spec__
|
|
* name
|
|
* origin
|
|
* ns - a copy (dict) of the module's __dict__ (or None)
|
|
* ns_id - the object ID of the module's __dict__
|
|
* cached - the sys.modules[mod.__spec__.name] entry (or None)
|
|
* cached_id - the object ID of the sys.modules entry (or None)
|
|
|
|
In cases where the value is not available (e.g. due to serialization),
|
|
the value will be None.
|
|
"""
|
|
_fields = tuple('id module ns ns_id cached cached_id'.split())
|
|
|
|
@classmethod
|
|
def from_module(cls, mod):
|
|
name = mod.__spec__.name
|
|
cached = sys.modules.get(name)
|
|
return cls(
|
|
id=id(mod),
|
|
module=mod,
|
|
ns=types.SimpleNamespace(**mod.__dict__),
|
|
ns_id=id(mod.__dict__),
|
|
cached=cached,
|
|
cached_id=id(cached),
|
|
)
|
|
|
|
SCRIPT = textwrap.dedent('''
|
|
{imports}
|
|
|
|
name = {name!r}
|
|
|
|
{prescript}
|
|
|
|
mod = {name}
|
|
|
|
{body}
|
|
|
|
{postscript}
|
|
''')
|
|
IMPORTS = textwrap.dedent('''
|
|
import sys
|
|
''').strip()
|
|
SCRIPT_BODY = textwrap.dedent('''
|
|
# Capture the snapshot data.
|
|
cached = sys.modules.get(name)
|
|
snapshot = dict(
|
|
id=id(mod),
|
|
module=dict(
|
|
__file__=mod.__file__,
|
|
__spec__=dict(
|
|
name=mod.__spec__.name,
|
|
origin=mod.__spec__.origin,
|
|
),
|
|
),
|
|
ns=None,
|
|
ns_id=id(mod.__dict__),
|
|
cached=None,
|
|
cached_id=id(cached) if cached else None,
|
|
)
|
|
''').strip()
|
|
CLEANUP_SCRIPT = textwrap.dedent('''
|
|
# Clean up the module.
|
|
sys.modules.pop(name, None)
|
|
''').strip()
|
|
|
|
@classmethod
|
|
def build_script(cls, name, *,
|
|
prescript=None,
|
|
import_first=False,
|
|
postscript=None,
|
|
postcleanup=False,
|
|
):
|
|
if postcleanup is True:
|
|
postcleanup = cls.CLEANUP_SCRIPT
|
|
elif isinstance(postcleanup, str):
|
|
postcleanup = textwrap.dedent(postcleanup).strip()
|
|
postcleanup = cls.CLEANUP_SCRIPT + os.linesep + postcleanup
|
|
else:
|
|
postcleanup = ''
|
|
prescript = textwrap.dedent(prescript).strip() if prescript else ''
|
|
postscript = textwrap.dedent(postscript).strip() if postscript else ''
|
|
|
|
if postcleanup:
|
|
if postscript:
|
|
postscript = postscript + os.linesep * 2 + postcleanup
|
|
else:
|
|
postscript = postcleanup
|
|
|
|
if import_first:
|
|
prescript += textwrap.dedent(f'''
|
|
|
|
# Now import the module.
|
|
assert name not in sys.modules
|
|
import {name}''')
|
|
|
|
return cls.SCRIPT.format(
|
|
imports=cls.IMPORTS.strip(),
|
|
name=name,
|
|
prescript=prescript.strip(),
|
|
body=cls.SCRIPT_BODY.strip(),
|
|
postscript=postscript,
|
|
)
|
|
|
|
@classmethod
|
|
def parse(cls, text):
|
|
raw = json.loads(text)
|
|
mod = raw['module']
|
|
mod['__spec__'] = types.SimpleNamespace(**mod['__spec__'])
|
|
raw['module'] = types.SimpleNamespace(**mod)
|
|
return cls(**raw)
|
|
|
|
@classmethod
|
|
def from_subinterp(cls, name, interpid=None, *, pipe=None, **script_kwds):
|
|
if pipe is not None:
|
|
return cls._from_subinterp(name, interpid, pipe, script_kwds)
|
|
pipe = os.pipe()
|
|
try:
|
|
return cls._from_subinterp(name, interpid, pipe, script_kwds)
|
|
finally:
|
|
r, w = pipe
|
|
os.close(r)
|
|
os.close(w)
|
|
|
|
@classmethod
|
|
def _from_subinterp(cls, name, interpid, pipe, script_kwargs):
|
|
r, w = pipe
|
|
|
|
# Build the script.
|
|
postscript = textwrap.dedent(f'''
|
|
# Send the result over the pipe.
|
|
import json
|
|
import os
|
|
os.write({w}, json.dumps(snapshot).encode())
|
|
|
|
''')
|
|
_postscript = script_kwargs.get('postscript')
|
|
if _postscript:
|
|
_postscript = textwrap.dedent(_postscript).lstrip()
|
|
postscript += _postscript
|
|
script_kwargs['postscript'] = postscript.strip()
|
|
script = cls.build_script(name, **script_kwargs)
|
|
|
|
# Run the script.
|
|
if interpid is None:
|
|
ret = support.run_in_subinterp(script)
|
|
if ret != 0:
|
|
raise AssertionError(f'{ret} != 0')
|
|
else:
|
|
_interpreters.run_string(interpid, script)
|
|
|
|
# Parse the results.
|
|
text = os.read(r, 1000)
|
|
return cls.parse(text.decode())
|
|
|
|
|
|
class LockTests(unittest.TestCase):
|
|
|
|
"""Very basic test of import lock functions."""
|
|
|
|
def verify_lock_state(self, expected):
|
|
self.assertEqual(imp.lock_held(), expected,
|
|
"expected imp.lock_held() to be %r" % expected)
|
|
def testLock(self):
|
|
LOOPS = 50
|
|
|
|
# The import lock may already be held, e.g. if the test suite is run
|
|
# via "import test.autotest".
|
|
lock_held_at_start = imp.lock_held()
|
|
self.verify_lock_state(lock_held_at_start)
|
|
|
|
for i in range(LOOPS):
|
|
imp.acquire_lock()
|
|
self.verify_lock_state(True)
|
|
|
|
for i in range(LOOPS):
|
|
imp.release_lock()
|
|
|
|
# The original state should be restored now.
|
|
self.verify_lock_state(lock_held_at_start)
|
|
|
|
if not lock_held_at_start:
|
|
try:
|
|
imp.release_lock()
|
|
except RuntimeError:
|
|
pass
|
|
else:
|
|
self.fail("release_lock() without lock should raise "
|
|
"RuntimeError")
|
|
|
|
class ImportTests(unittest.TestCase):
|
|
def setUp(self):
|
|
mod = importlib.import_module('test.encoded_modules')
|
|
self.test_strings = mod.test_strings
|
|
self.test_path = mod.__path__
|
|
|
|
# test_import_encoded_module moved to test_source_encoding.py
|
|
|
|
def test_find_module_encoding(self):
|
|
for mod, encoding, _ in self.test_strings:
|
|
with imp.find_module('module_' + mod, self.test_path)[0] as fd:
|
|
self.assertEqual(fd.encoding, encoding)
|
|
|
|
path = [os.path.dirname(__file__)]
|
|
with self.assertRaises(SyntaxError):
|
|
imp.find_module('badsyntax_pep3120', path)
|
|
|
|
def test_issue1267(self):
|
|
for mod, encoding, _ in self.test_strings:
|
|
fp, filename, info = imp.find_module('module_' + mod,
|
|
self.test_path)
|
|
with fp:
|
|
self.assertNotEqual(fp, None)
|
|
self.assertEqual(fp.encoding, encoding)
|
|
self.assertEqual(fp.tell(), 0)
|
|
self.assertEqual(fp.readline(), '# test %s encoding\n'
|
|
% encoding)
|
|
|
|
fp, filename, info = imp.find_module("tokenize")
|
|
with fp:
|
|
self.assertNotEqual(fp, None)
|
|
self.assertEqual(fp.encoding, "utf-8")
|
|
self.assertEqual(fp.tell(), 0)
|
|
self.assertEqual(fp.readline(),
|
|
'"""Tokenization help for Python programs.\n')
|
|
|
|
def test_issue3594(self):
|
|
temp_mod_name = 'test_imp_helper'
|
|
sys.path.insert(0, '.')
|
|
try:
|
|
with open(temp_mod_name + '.py', 'w', encoding="latin-1") as file:
|
|
file.write("# coding: cp1252\nu = 'test.test_imp'\n")
|
|
file, filename, info = imp.find_module(temp_mod_name)
|
|
file.close()
|
|
self.assertEqual(file.encoding, 'cp1252')
|
|
finally:
|
|
del sys.path[0]
|
|
os_helper.unlink(temp_mod_name + '.py')
|
|
os_helper.unlink(temp_mod_name + '.pyc')
|
|
|
|
def test_issue5604(self):
|
|
# Test cannot cover imp.load_compiled function.
|
|
# Martin von Loewis note what shared library cannot have non-ascii
|
|
# character because init_xxx function cannot be compiled
|
|
# and issue never happens for dynamic modules.
|
|
# But sources modified to follow generic way for processing paths.
|
|
|
|
# the return encoding could be uppercase or None
|
|
fs_encoding = sys.getfilesystemencoding()
|
|
|
|
# covers utf-8 and Windows ANSI code pages
|
|
# one non-space symbol from every page
|
|
# (http://en.wikipedia.org/wiki/Code_page)
|
|
known_locales = {
|
|
'utf-8' : b'\xc3\xa4',
|
|
'cp1250' : b'\x8C',
|
|
'cp1251' : b'\xc0',
|
|
'cp1252' : b'\xc0',
|
|
'cp1253' : b'\xc1',
|
|
'cp1254' : b'\xc0',
|
|
'cp1255' : b'\xe0',
|
|
'cp1256' : b'\xe0',
|
|
'cp1257' : b'\xc0',
|
|
'cp1258' : b'\xc0',
|
|
}
|
|
|
|
if sys.platform == 'darwin':
|
|
self.assertEqual(fs_encoding, 'utf-8')
|
|
# Mac OS X uses the Normal Form D decomposition
|
|
# http://developer.apple.com/mac/library/qa/qa2001/qa1173.html
|
|
special_char = b'a\xcc\x88'
|
|
else:
|
|
special_char = known_locales.get(fs_encoding)
|
|
|
|
if not special_char:
|
|
self.skipTest("can't run this test with %s as filesystem encoding"
|
|
% fs_encoding)
|
|
decoded_char = special_char.decode(fs_encoding)
|
|
temp_mod_name = 'test_imp_helper_' + decoded_char
|
|
test_package_name = 'test_imp_helper_package_' + decoded_char
|
|
init_file_name = os.path.join(test_package_name, '__init__.py')
|
|
try:
|
|
# if the curdir is not in sys.path the test fails when run with
|
|
# ./python ./Lib/test/regrtest.py test_imp
|
|
sys.path.insert(0, os.curdir)
|
|
with open(temp_mod_name + '.py', 'w', encoding="utf-8") as file:
|
|
file.write('a = 1\n')
|
|
file, filename, info = imp.find_module(temp_mod_name)
|
|
with file:
|
|
self.assertIsNotNone(file)
|
|
self.assertTrue(filename[:-3].endswith(temp_mod_name))
|
|
self.assertEqual(info[0], '.py')
|
|
self.assertEqual(info[1], 'r')
|
|
self.assertEqual(info[2], imp.PY_SOURCE)
|
|
|
|
mod = imp.load_module(temp_mod_name, file, filename, info)
|
|
self.assertEqual(mod.a, 1)
|
|
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter('ignore')
|
|
mod = imp.load_source(temp_mod_name, temp_mod_name + '.py')
|
|
self.assertEqual(mod.a, 1)
|
|
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter('ignore')
|
|
if not sys.dont_write_bytecode:
|
|
mod = imp.load_compiled(
|
|
temp_mod_name,
|
|
imp.cache_from_source(temp_mod_name + '.py'))
|
|
self.assertEqual(mod.a, 1)
|
|
|
|
if not os.path.exists(test_package_name):
|
|
os.mkdir(test_package_name)
|
|
with open(init_file_name, 'w', encoding="utf-8") as file:
|
|
file.write('b = 2\n')
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter('ignore')
|
|
package = imp.load_package(test_package_name, test_package_name)
|
|
self.assertEqual(package.b, 2)
|
|
finally:
|
|
del sys.path[0]
|
|
for ext in ('.py', '.pyc'):
|
|
os_helper.unlink(temp_mod_name + ext)
|
|
os_helper.unlink(init_file_name + ext)
|
|
os_helper.rmtree(test_package_name)
|
|
os_helper.rmtree('__pycache__')
|
|
|
|
def test_issue9319(self):
|
|
path = os.path.dirname(__file__)
|
|
self.assertRaises(SyntaxError,
|
|
imp.find_module, "badsyntax_pep3120", [path])
|
|
|
|
def test_load_from_source(self):
|
|
# Verify that the imp module can correctly load and find .py files
|
|
# XXX (ncoghlan): It would be nice to use import_helper.CleanImport
|
|
# here, but that breaks because the os module registers some
|
|
# handlers in copy_reg on import. Since CleanImport doesn't
|
|
# revert that registration, the module is left in a broken
|
|
# state after reversion. Reinitialising the module contents
|
|
# and just reverting os.environ to its previous state is an OK
|
|
# workaround
|
|
with import_helper.CleanImport('os', 'os.path', OS_PATH_NAME):
|
|
import os
|
|
orig_path = os.path
|
|
orig_getenv = os.getenv
|
|
with os_helper.EnvironmentVarGuard():
|
|
x = imp.find_module("os")
|
|
self.addCleanup(x[0].close)
|
|
new_os = imp.load_module("os", *x)
|
|
self.assertIs(os, new_os)
|
|
self.assertIs(orig_path, new_os.path)
|
|
self.assertIsNot(orig_getenv, new_os.getenv)
|
|
|
|
@requires_load_dynamic
|
|
def test_issue15828_load_extensions(self):
|
|
# Issue 15828 picked up that the adapter between the old imp API
|
|
# and importlib couldn't handle C extensions
|
|
example = "_heapq"
|
|
x = imp.find_module(example)
|
|
file_ = x[0]
|
|
if file_ is not None:
|
|
self.addCleanup(file_.close)
|
|
mod = imp.load_module(example, *x)
|
|
self.assertEqual(mod.__name__, example)
|
|
|
|
@requires_load_dynamic
|
|
def test_issue16421_multiple_modules_in_one_dll(self):
|
|
# Issue 16421: loading several modules from the same compiled file fails
|
|
m = '_testimportmultiple'
|
|
fileobj, pathname, description = imp.find_module(m)
|
|
fileobj.close()
|
|
mod0 = imp.load_dynamic(m, pathname)
|
|
mod1 = imp.load_dynamic('_testimportmultiple_foo', pathname)
|
|
mod2 = imp.load_dynamic('_testimportmultiple_bar', pathname)
|
|
self.assertEqual(mod0.__name__, m)
|
|
self.assertEqual(mod1.__name__, '_testimportmultiple_foo')
|
|
self.assertEqual(mod2.__name__, '_testimportmultiple_bar')
|
|
with self.assertRaises(ImportError):
|
|
imp.load_dynamic('nonexistent', pathname)
|
|
|
|
@requires_load_dynamic
|
|
def test_load_dynamic_ImportError_path(self):
|
|
# Issue #1559549 added `name` and `path` attributes to ImportError
|
|
# in order to provide better detail. Issue #10854 implemented those
|
|
# attributes on import failures of extensions on Windows.
|
|
path = 'bogus file path'
|
|
name = 'extension'
|
|
with self.assertRaises(ImportError) as err:
|
|
imp.load_dynamic(name, path)
|
|
self.assertIn(path, err.exception.path)
|
|
self.assertEqual(name, err.exception.name)
|
|
|
|
@requires_load_dynamic
|
|
def test_load_module_extension_file_is_None(self):
|
|
# When loading an extension module and the file is None, open one
|
|
# on the behalf of imp.load_dynamic().
|
|
# Issue #15902
|
|
name = '_testimportmultiple'
|
|
found = imp.find_module(name)
|
|
if found[0] is not None:
|
|
found[0].close()
|
|
if found[2][2] != imp.C_EXTENSION:
|
|
self.skipTest("found module doesn't appear to be a C extension")
|
|
imp.load_module(name, None, *found[1:])
|
|
|
|
@requires_load_dynamic
|
|
def test_issue24748_load_module_skips_sys_modules_check(self):
|
|
name = 'test.imp_dummy'
|
|
try:
|
|
del sys.modules[name]
|
|
except KeyError:
|
|
pass
|
|
try:
|
|
module = importlib.import_module(name)
|
|
spec = importlib.util.find_spec('_testmultiphase')
|
|
module = imp.load_dynamic(name, spec.origin)
|
|
self.assertEqual(module.__name__, name)
|
|
self.assertEqual(module.__spec__.name, name)
|
|
self.assertEqual(module.__spec__.origin, spec.origin)
|
|
self.assertRaises(AttributeError, getattr, module, 'dummy_name')
|
|
self.assertEqual(module.int_const, 1969)
|
|
self.assertIs(sys.modules[name], module)
|
|
finally:
|
|
try:
|
|
del sys.modules[name]
|
|
except KeyError:
|
|
pass
|
|
|
|
@unittest.skipIf(sys.dont_write_bytecode,
|
|
"test meaningful only when writing bytecode")
|
|
def test_bug7732(self):
|
|
with os_helper.temp_cwd():
|
|
source = os_helper.TESTFN + '.py'
|
|
os.mkdir(source)
|
|
self.assertRaisesRegex(ImportError, '^No module',
|
|
imp.find_module, os_helper.TESTFN, ["."])
|
|
|
|
def test_multiple_calls_to_get_data(self):
|
|
# Issue #18755: make sure multiple calls to get_data() can succeed.
|
|
loader = imp._LoadSourceCompatibility('imp', imp.__file__,
|
|
open(imp.__file__, encoding="utf-8"))
|
|
loader.get_data(imp.__file__) # File should be closed
|
|
loader.get_data(imp.__file__) # Will need to create a newly opened file
|
|
|
|
def test_load_source(self):
|
|
# Create a temporary module since load_source(name) modifies
|
|
# sys.modules[name] attributes like __loader___
|
|
modname = f"tmp{__name__}"
|
|
mod = type(sys.modules[__name__])(modname)
|
|
with support.swap_item(sys.modules, modname, mod):
|
|
with self.assertRaisesRegex(ValueError, 'embedded null'):
|
|
imp.load_source(modname, __file__ + "\0")
|
|
|
|
@support.cpython_only
|
|
def test_issue31315(self):
|
|
# There shouldn't be an assertion failure in imp.create_dynamic(),
|
|
# when spec.name is not a string.
|
|
create_dynamic = support.get_attribute(imp, 'create_dynamic')
|
|
class BadSpec:
|
|
name = None
|
|
origin = 'foo'
|
|
with self.assertRaises(TypeError):
|
|
create_dynamic(BadSpec())
|
|
|
|
def test_issue_35321(self):
|
|
# Both _frozen_importlib and _frozen_importlib_external
|
|
# should have a spec origin of "frozen" and
|
|
# no need to clean up imports in this case.
|
|
|
|
import _frozen_importlib_external
|
|
self.assertEqual(_frozen_importlib_external.__spec__.origin, "frozen")
|
|
|
|
import _frozen_importlib
|
|
self.assertEqual(_frozen_importlib.__spec__.origin, "frozen")
|
|
|
|
def test_source_hash(self):
|
|
self.assertEqual(_imp.source_hash(42, b'hi'), b'\xfb\xd9G\x05\xaf$\x9b~')
|
|
self.assertEqual(_imp.source_hash(43, b'hi'), b'\xd0/\x87C\xccC\xff\xe2')
|
|
|
|
def test_pyc_invalidation_mode_from_cmdline(self):
|
|
cases = [
|
|
([], "default"),
|
|
(["--check-hash-based-pycs", "default"], "default"),
|
|
(["--check-hash-based-pycs", "always"], "always"),
|
|
(["--check-hash-based-pycs", "never"], "never"),
|
|
]
|
|
for interp_args, expected in cases:
|
|
args = interp_args + [
|
|
"-c",
|
|
"import _imp; print(_imp.check_hash_based_pycs)",
|
|
]
|
|
res = script_helper.assert_python_ok(*args)
|
|
self.assertEqual(res.out.strip().decode('utf-8'), expected)
|
|
|
|
def test_find_and_load_checked_pyc(self):
|
|
# issue 34056
|
|
with os_helper.temp_cwd():
|
|
with open('mymod.py', 'wb') as fp:
|
|
fp.write(b'x = 42\n')
|
|
py_compile.compile(
|
|
'mymod.py',
|
|
doraise=True,
|
|
invalidation_mode=py_compile.PycInvalidationMode.CHECKED_HASH,
|
|
)
|
|
file, path, description = imp.find_module('mymod', path=['.'])
|
|
mod = imp.load_module('mymod', file, path, description)
|
|
self.assertEqual(mod.x, 42)
|
|
|
|
def test_issue98354(self):
|
|
# _imp.create_builtin should raise TypeError
|
|
# if 'name' attribute of 'spec' argument is not a 'str' instance
|
|
|
|
create_builtin = support.get_attribute(_imp, "create_builtin")
|
|
|
|
class FakeSpec:
|
|
def __init__(self, name):
|
|
self.name = self
|
|
spec = FakeSpec("time")
|
|
with self.assertRaises(TypeError):
|
|
create_builtin(spec)
|
|
|
|
class FakeSpec2:
|
|
name = [1, 2, 3, 4]
|
|
spec = FakeSpec2()
|
|
with self.assertRaises(TypeError):
|
|
create_builtin(spec)
|
|
|
|
import builtins
|
|
class UnicodeSubclass(str):
|
|
pass
|
|
class GoodSpec:
|
|
name = UnicodeSubclass("builtins")
|
|
spec = GoodSpec()
|
|
bltin = create_builtin(spec)
|
|
self.assertEqual(bltin, builtins)
|
|
|
|
class UnicodeSubclassFakeSpec(str):
|
|
def __init__(self, name):
|
|
self.name = self
|
|
spec = UnicodeSubclassFakeSpec("builtins")
|
|
bltin = create_builtin(spec)
|
|
self.assertEqual(bltin, builtins)
|
|
|
|
@support.cpython_only
|
|
def test_create_builtin_subinterp(self):
|
|
# gh-99578: create_builtin() behavior changes after the creation of the
|
|
# first sub-interpreter. Test both code paths, before and after the
|
|
# creation of a sub-interpreter. Previously, create_builtin() had
|
|
# a reference leak after the creation of the first sub-interpreter.
|
|
|
|
import builtins
|
|
create_builtin = support.get_attribute(_imp, "create_builtin")
|
|
class Spec:
|
|
name = "builtins"
|
|
spec = Spec()
|
|
|
|
def check_get_builtins():
|
|
refcnt = sys.getrefcount(builtins)
|
|
mod = _imp.create_builtin(spec)
|
|
self.assertIs(mod, builtins)
|
|
self.assertEqual(sys.getrefcount(builtins), refcnt + 1)
|
|
# Check that a GC collection doesn't crash
|
|
gc.collect()
|
|
|
|
check_get_builtins()
|
|
|
|
ret = support.run_in_subinterp("import builtins")
|
|
self.assertEqual(ret, 0)
|
|
|
|
check_get_builtins()
|
|
|
|
|
|
class TestSinglePhaseSnapshot(ModuleSnapshot):
|
|
|
|
@classmethod
|
|
def from_module(cls, mod):
|
|
self = super().from_module(mod)
|
|
self.summed = mod.sum(1, 2)
|
|
self.lookedup = mod.look_up_self()
|
|
self.lookedup_id = id(self.lookedup)
|
|
self.state_initialized = mod.state_initialized()
|
|
if hasattr(mod, 'initialized_count'):
|
|
self.init_count = mod.initialized_count()
|
|
return self
|
|
|
|
SCRIPT_BODY = ModuleSnapshot.SCRIPT_BODY + textwrap.dedent(f'''
|
|
snapshot['module'].update(dict(
|
|
int_const=mod.int_const,
|
|
str_const=mod.str_const,
|
|
_module_initialized=mod._module_initialized,
|
|
))
|
|
snapshot.update(dict(
|
|
summed=mod.sum(1, 2),
|
|
lookedup_id=id(mod.look_up_self()),
|
|
state_initialized=mod.state_initialized(),
|
|
init_count=mod.initialized_count(),
|
|
has_spam=hasattr(mod, 'spam'),
|
|
spam=getattr(mod, 'spam', None),
|
|
))
|
|
''').rstrip()
|
|
|
|
@classmethod
|
|
def parse(cls, text):
|
|
self = super().parse(text)
|
|
if not self.has_spam:
|
|
del self.spam
|
|
del self.has_spam
|
|
return self
|
|
|
|
|
|
@requires_load_dynamic
|
|
class SinglephaseInitTests(unittest.TestCase):
|
|
|
|
NAME = '_testsinglephase'
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
if '-R' in sys.argv or '--huntrleaks' in sys.argv:
|
|
# https://github.com/python/cpython/issues/102251
|
|
raise unittest.SkipTest('unresolved refleaks (see gh-102251)')
|
|
fileobj, filename, _ = imp.find_module(cls.NAME)
|
|
fileobj.close()
|
|
cls.FILE = filename
|
|
|
|
# Start fresh.
|
|
cls.clean_up()
|
|
|
|
def tearDown(self):
|
|
# Clean up the module.
|
|
self.clean_up()
|
|
|
|
@classmethod
|
|
def clean_up(cls):
|
|
name = cls.NAME
|
|
filename = cls.FILE
|
|
if name in sys.modules:
|
|
if hasattr(sys.modules[name], '_clear_globals'):
|
|
assert sys.modules[name].__file__ == filename
|
|
sys.modules[name]._clear_globals()
|
|
del sys.modules[name]
|
|
# Clear all internally cached data for the extension.
|
|
_testinternalcapi.clear_extension(name, filename)
|
|
|
|
#########################
|
|
# helpers
|
|
|
|
def add_module_cleanup(self, name):
|
|
def clean_up():
|
|
# Clear all internally cached data for the extension.
|
|
_testinternalcapi.clear_extension(name, self.FILE)
|
|
self.addCleanup(clean_up)
|
|
|
|
def load(self, name):
|
|
try:
|
|
already_loaded = self.already_loaded
|
|
except AttributeError:
|
|
already_loaded = self.already_loaded = {}
|
|
assert name not in already_loaded
|
|
mod = imp.load_dynamic(name, self.FILE)
|
|
self.assertNotIn(mod, already_loaded.values())
|
|
already_loaded[name] = mod
|
|
return types.SimpleNamespace(
|
|
name=name,
|
|
module=mod,
|
|
snapshot=TestSinglePhaseSnapshot.from_module(mod),
|
|
)
|
|
|
|
def re_load(self, name, mod):
|
|
assert sys.modules[name] is mod
|
|
assert mod.__dict__ == mod.__dict__
|
|
reloaded = imp.load_dynamic(name, self.FILE)
|
|
return types.SimpleNamespace(
|
|
name=name,
|
|
module=reloaded,
|
|
snapshot=TestSinglePhaseSnapshot.from_module(reloaded),
|
|
)
|
|
|
|
# subinterpreters
|
|
|
|
def add_subinterpreter(self):
|
|
interpid = _interpreters.create(isolated=False)
|
|
_interpreters.run_string(interpid, textwrap.dedent('''
|
|
import sys
|
|
import _testinternalcapi
|
|
'''))
|
|
def clean_up():
|
|
_interpreters.run_string(interpid, textwrap.dedent(f'''
|
|
name = {self.NAME!r}
|
|
if name in sys.modules:
|
|
sys.modules[name]._clear_globals()
|
|
_testinternalcapi.clear_extension(name, {self.FILE!r})
|
|
'''))
|
|
_interpreters.destroy(interpid)
|
|
self.addCleanup(clean_up)
|
|
return interpid
|
|
|
|
def import_in_subinterp(self, interpid=None, *,
|
|
postscript=None,
|
|
postcleanup=False,
|
|
):
|
|
name = self.NAME
|
|
|
|
if postcleanup:
|
|
import_ = 'import _testinternalcapi' if interpid is None else ''
|
|
postcleanup = f'''
|
|
{import_}
|
|
mod._clear_globals()
|
|
_testinternalcapi.clear_extension(name, {self.FILE!r})
|
|
'''
|
|
|
|
try:
|
|
pipe = self._pipe
|
|
except AttributeError:
|
|
r, w = pipe = self._pipe = os.pipe()
|
|
self.addCleanup(os.close, r)
|
|
self.addCleanup(os.close, w)
|
|
|
|
snapshot = TestSinglePhaseSnapshot.from_subinterp(
|
|
name,
|
|
interpid,
|
|
pipe=pipe,
|
|
import_first=True,
|
|
postscript=postscript,
|
|
postcleanup=postcleanup,
|
|
)
|
|
|
|
return types.SimpleNamespace(
|
|
name=name,
|
|
module=None,
|
|
snapshot=snapshot,
|
|
)
|
|
|
|
# checks
|
|
|
|
def check_common(self, loaded):
|
|
isolated = False
|
|
|
|
mod = loaded.module
|
|
if not mod:
|
|
# It came from a subinterpreter.
|
|
isolated = True
|
|
mod = loaded.snapshot.module
|
|
# mod.__name__ might not match, but the spec will.
|
|
self.assertEqual(mod.__spec__.name, loaded.name)
|
|
self.assertEqual(mod.__file__, self.FILE)
|
|
self.assertEqual(mod.__spec__.origin, self.FILE)
|
|
if not isolated:
|
|
self.assertTrue(issubclass(mod.error, Exception))
|
|
self.assertEqual(mod.int_const, 1969)
|
|
self.assertEqual(mod.str_const, 'something different')
|
|
self.assertIsInstance(mod._module_initialized, float)
|
|
self.assertGreater(mod._module_initialized, 0)
|
|
|
|
snap = loaded.snapshot
|
|
self.assertEqual(snap.summed, 3)
|
|
if snap.state_initialized is not None:
|
|
self.assertIsInstance(snap.state_initialized, float)
|
|
self.assertGreater(snap.state_initialized, 0)
|
|
if isolated:
|
|
# The "looked up" module is interpreter-specific
|
|
# (interp->imports.modules_by_index was set for the module).
|
|
self.assertEqual(snap.lookedup_id, snap.id)
|
|
self.assertEqual(snap.cached_id, snap.id)
|
|
with self.assertRaises(AttributeError):
|
|
snap.spam
|
|
else:
|
|
self.assertIs(snap.lookedup, mod)
|
|
self.assertIs(snap.cached, mod)
|
|
|
|
def check_direct(self, loaded):
|
|
# The module has its own PyModuleDef, with a matching name.
|
|
self.assertEqual(loaded.module.__name__, loaded.name)
|
|
self.assertIs(loaded.snapshot.lookedup, loaded.module)
|
|
|
|
def check_indirect(self, loaded, orig):
|
|
# The module re-uses another's PyModuleDef, with a different name.
|
|
assert orig is not loaded.module
|
|
assert orig.__name__ != loaded.name
|
|
self.assertNotEqual(loaded.module.__name__, loaded.name)
|
|
self.assertIs(loaded.snapshot.lookedup, loaded.module)
|
|
|
|
def check_basic(self, loaded, expected_init_count):
|
|
# m_size == -1
|
|
# The module loads fresh the first time and copies m_copy after.
|
|
snap = loaded.snapshot
|
|
self.assertIsNot(snap.state_initialized, None)
|
|
self.assertIsInstance(snap.init_count, int)
|
|
self.assertGreater(snap.init_count, 0)
|
|
self.assertEqual(snap.init_count, expected_init_count)
|
|
|
|
def check_with_reinit(self, loaded):
|
|
# m_size >= 0
|
|
# The module loads fresh every time.
|
|
pass
|
|
|
|
def check_fresh(self, loaded):
|
|
"""
|
|
The module had not been loaded before (at least since fully reset).
|
|
"""
|
|
snap = loaded.snapshot
|
|
# The module's init func was run.
|
|
# A copy of the module's __dict__ was stored in def->m_base.m_copy.
|
|
# The previous m_copy was deleted first.
|
|
# _PyRuntime.imports.extensions was set.
|
|
self.assertEqual(snap.init_count, 1)
|
|
# The global state was initialized.
|
|
# The module attrs were initialized from that state.
|
|
self.assertEqual(snap.module._module_initialized,
|
|
snap.state_initialized)
|
|
|
|
def check_semi_fresh(self, loaded, base, prev):
|
|
"""
|
|
The module had been loaded before and then reset
|
|
(but the module global state wasn't).
|
|
"""
|
|
snap = loaded.snapshot
|
|
# The module's init func was run again.
|
|
# A copy of the module's __dict__ was stored in def->m_base.m_copy.
|
|
# The previous m_copy was deleted first.
|
|
# The module globals did not get reset.
|
|
self.assertNotEqual(snap.id, base.snapshot.id)
|
|
self.assertNotEqual(snap.id, prev.snapshot.id)
|
|
self.assertEqual(snap.init_count, prev.snapshot.init_count + 1)
|
|
# The global state was updated.
|
|
# The module attrs were initialized from that state.
|
|
self.assertEqual(snap.module._module_initialized,
|
|
snap.state_initialized)
|
|
self.assertNotEqual(snap.state_initialized,
|
|
base.snapshot.state_initialized)
|
|
self.assertNotEqual(snap.state_initialized,
|
|
prev.snapshot.state_initialized)
|
|
|
|
def check_copied(self, loaded, base):
|
|
"""
|
|
The module had been loaded before and never reset.
|
|
"""
|
|
snap = loaded.snapshot
|
|
# The module's init func was not run again.
|
|
# The interpreter copied m_copy, as set by the other interpreter,
|
|
# with objects owned by the other interpreter.
|
|
# The module globals did not get reset.
|
|
self.assertNotEqual(snap.id, base.snapshot.id)
|
|
self.assertEqual(snap.init_count, base.snapshot.init_count)
|
|
# The global state was not updated since the init func did not run.
|
|
# The module attrs were not directly initialized from that state.
|
|
# The state and module attrs still match the previous loading.
|
|
self.assertEqual(snap.module._module_initialized,
|
|
snap.state_initialized)
|
|
self.assertEqual(snap.state_initialized,
|
|
base.snapshot.state_initialized)
|
|
|
|
#########################
|
|
# the tests
|
|
|
|
def test_cleared_globals(self):
|
|
loaded = self.load(self.NAME)
|
|
_testsinglephase = loaded.module
|
|
init_before = _testsinglephase.state_initialized()
|
|
|
|
_testsinglephase._clear_globals()
|
|
init_after = _testsinglephase.state_initialized()
|
|
init_count = _testsinglephase.initialized_count()
|
|
|
|
self.assertGreater(init_before, 0)
|
|
self.assertEqual(init_after, 0)
|
|
self.assertEqual(init_count, -1)
|
|
|
|
def test_variants(self):
|
|
# Exercise the most meaningful variants described in Python/import.c.
|
|
self.maxDiff = None
|
|
|
|
# Check the "basic" module.
|
|
|
|
name = self.NAME
|
|
expected_init_count = 1
|
|
with self.subTest(name):
|
|
loaded = self.load(name)
|
|
|
|
self.check_common(loaded)
|
|
self.check_direct(loaded)
|
|
self.check_basic(loaded, expected_init_count)
|
|
basic = loaded.module
|
|
|
|
# Check its indirect variants.
|
|
|
|
name = f'{self.NAME}_basic_wrapper'
|
|
self.add_module_cleanup(name)
|
|
expected_init_count += 1
|
|
with self.subTest(name):
|
|
loaded = self.load(name)
|
|
|
|
self.check_common(loaded)
|
|
self.check_indirect(loaded, basic)
|
|
self.check_basic(loaded, expected_init_count)
|
|
|
|
# Currently PyState_AddModule() always replaces the cached module.
|
|
self.assertIs(basic.look_up_self(), loaded.module)
|
|
self.assertEqual(basic.initialized_count(), expected_init_count)
|
|
|
|
# The cached module shouldn't change after this point.
|
|
basic_lookedup = loaded.module
|
|
|
|
# Check its direct variant.
|
|
|
|
name = f'{self.NAME}_basic_copy'
|
|
self.add_module_cleanup(name)
|
|
expected_init_count += 1
|
|
with self.subTest(name):
|
|
loaded = self.load(name)
|
|
|
|
self.check_common(loaded)
|
|
self.check_direct(loaded)
|
|
self.check_basic(loaded, expected_init_count)
|
|
|
|
# This should change the cached module for _testsinglephase.
|
|
self.assertIs(basic.look_up_self(), basic_lookedup)
|
|
self.assertEqual(basic.initialized_count(), expected_init_count)
|
|
|
|
# Check the non-basic variant that has no state.
|
|
|
|
name = f'{self.NAME}_with_reinit'
|
|
self.add_module_cleanup(name)
|
|
with self.subTest(name):
|
|
loaded = self.load(name)
|
|
|
|
self.check_common(loaded)
|
|
self.assertIs(loaded.snapshot.state_initialized, None)
|
|
self.check_direct(loaded)
|
|
self.check_with_reinit(loaded)
|
|
|
|
# This should change the cached module for _testsinglephase.
|
|
self.assertIs(basic.look_up_self(), basic_lookedup)
|
|
self.assertEqual(basic.initialized_count(), expected_init_count)
|
|
|
|
# Check the basic variant that has state.
|
|
|
|
name = f'{self.NAME}_with_state'
|
|
self.add_module_cleanup(name)
|
|
with self.subTest(name):
|
|
loaded = self.load(name)
|
|
|
|
self.check_common(loaded)
|
|
self.assertIsNot(loaded.snapshot.state_initialized, None)
|
|
self.check_direct(loaded)
|
|
self.check_with_reinit(loaded)
|
|
|
|
# This should change the cached module for _testsinglephase.
|
|
self.assertIs(basic.look_up_self(), basic_lookedup)
|
|
self.assertEqual(basic.initialized_count(), expected_init_count)
|
|
|
|
def test_basic_reloaded(self):
|
|
# m_copy is copied into the existing module object.
|
|
# Global state is not changed.
|
|
self.maxDiff = None
|
|
|
|
for name in [
|
|
self.NAME, # the "basic" module
|
|
f'{self.NAME}_basic_wrapper', # the indirect variant
|
|
f'{self.NAME}_basic_copy', # the direct variant
|
|
]:
|
|
self.add_module_cleanup(name)
|
|
with self.subTest(name):
|
|
loaded = self.load(name)
|
|
reloaded = self.re_load(name, loaded.module)
|
|
|
|
self.check_common(loaded)
|
|
self.check_common(reloaded)
|
|
|
|
# Make sure the original __dict__ did not get replaced.
|
|
self.assertEqual(id(loaded.module.__dict__),
|
|
loaded.snapshot.ns_id)
|
|
self.assertEqual(loaded.snapshot.ns.__dict__,
|
|
loaded.module.__dict__)
|
|
|
|
self.assertEqual(reloaded.module.__spec__.name, reloaded.name)
|
|
self.assertEqual(reloaded.module.__name__,
|
|
reloaded.snapshot.ns.__name__)
|
|
|
|
self.assertIs(reloaded.module, loaded.module)
|
|
self.assertIs(reloaded.module.__dict__, loaded.module.__dict__)
|
|
# It only happens to be the same but that's good enough here.
|
|
# We really just want to verify that the re-loaded attrs
|
|
# didn't change.
|
|
self.assertIs(reloaded.snapshot.lookedup,
|
|
loaded.snapshot.lookedup)
|
|
self.assertEqual(reloaded.snapshot.state_initialized,
|
|
loaded.snapshot.state_initialized)
|
|
self.assertEqual(reloaded.snapshot.init_count,
|
|
loaded.snapshot.init_count)
|
|
|
|
self.assertIs(reloaded.snapshot.cached, reloaded.module)
|
|
|
|
def test_with_reinit_reloaded(self):
|
|
# The module's m_init func is run again.
|
|
self.maxDiff = None
|
|
|
|
# Keep a reference around.
|
|
basic = self.load(self.NAME)
|
|
|
|
for name in [
|
|
f'{self.NAME}_with_reinit', # m_size == 0
|
|
f'{self.NAME}_with_state', # m_size > 0
|
|
]:
|
|
self.add_module_cleanup(name)
|
|
with self.subTest(name):
|
|
loaded = self.load(name)
|
|
reloaded = self.re_load(name, loaded.module)
|
|
|
|
self.check_common(loaded)
|
|
self.check_common(reloaded)
|
|
|
|
# Make sure the original __dict__ did not get replaced.
|
|
self.assertEqual(id(loaded.module.__dict__),
|
|
loaded.snapshot.ns_id)
|
|
self.assertEqual(loaded.snapshot.ns.__dict__,
|
|
loaded.module.__dict__)
|
|
|
|
self.assertEqual(reloaded.module.__spec__.name, reloaded.name)
|
|
self.assertEqual(reloaded.module.__name__,
|
|
reloaded.snapshot.ns.__name__)
|
|
|
|
self.assertIsNot(reloaded.module, loaded.module)
|
|
self.assertNotEqual(reloaded.module.__dict__,
|
|
loaded.module.__dict__)
|
|
self.assertIs(reloaded.snapshot.lookedup, reloaded.module)
|
|
if loaded.snapshot.state_initialized is None:
|
|
self.assertIs(reloaded.snapshot.state_initialized, None)
|
|
else:
|
|
self.assertGreater(reloaded.snapshot.state_initialized,
|
|
loaded.snapshot.state_initialized)
|
|
|
|
self.assertIs(reloaded.snapshot.cached, reloaded.module)
|
|
|
|
# Currently, for every single-phrase init module loaded
|
|
# in multiple interpreters, those interpreters share a
|
|
# PyModuleDef for that object, which can be a problem.
|
|
# Also, we test with a single-phase module that has global state,
|
|
# which is shared by all interpreters.
|
|
|
|
@requires_subinterpreters
|
|
def test_basic_multiple_interpreters_main_no_reset(self):
|
|
# without resetting; already loaded in main interpreter
|
|
|
|
# At this point:
|
|
# * alive in 0 interpreters
|
|
# * module def may or may not be loaded already
|
|
# * module def not in _PyRuntime.imports.extensions
|
|
# * mod init func has not run yet (since reset, at least)
|
|
# * m_copy not set (hasn't been loaded yet or already cleared)
|
|
# * module's global state has not been initialized yet
|
|
# (or already cleared)
|
|
|
|
main_loaded = self.load(self.NAME)
|
|
_testsinglephase = main_loaded.module
|
|
# Attrs set after loading are not in m_copy.
|
|
_testsinglephase.spam = 'spam, spam, spam, spam, eggs, and spam'
|
|
|
|
self.check_common(main_loaded)
|
|
self.check_fresh(main_loaded)
|
|
|
|
interpid1 = self.add_subinterpreter()
|
|
interpid2 = self.add_subinterpreter()
|
|
|
|
# At this point:
|
|
# * alive in 1 interpreter (main)
|
|
# * module def in _PyRuntime.imports.extensions
|
|
# * mod init func ran for the first time (since reset, at least)
|
|
# * m_copy was copied from the main interpreter (was NULL)
|
|
# * module's global state was initialized
|
|
|
|
# Use an interpreter that gets destroyed right away.
|
|
loaded = self.import_in_subinterp()
|
|
self.check_common(loaded)
|
|
self.check_copied(loaded, main_loaded)
|
|
|
|
# At this point:
|
|
# * alive in 1 interpreter (main)
|
|
# * module def still in _PyRuntime.imports.extensions
|
|
# * mod init func ran again
|
|
# * m_copy is NULL (claered when the interpreter was destroyed)
|
|
# (was from main interpreter)
|
|
# * module's global state was updated, not reset
|
|
|
|
# Use a subinterpreter that sticks around.
|
|
loaded = self.import_in_subinterp(interpid1)
|
|
self.check_common(loaded)
|
|
self.check_copied(loaded, main_loaded)
|
|
|
|
# At this point:
|
|
# * alive in 2 interpreters (main, interp1)
|
|
# * module def still in _PyRuntime.imports.extensions
|
|
# * mod init func ran again
|
|
# * m_copy was copied from interp1
|
|
# * module's global state was updated, not reset
|
|
|
|
# Use a subinterpreter while the previous one is still alive.
|
|
loaded = self.import_in_subinterp(interpid2)
|
|
self.check_common(loaded)
|
|
self.check_copied(loaded, main_loaded)
|
|
|
|
# At this point:
|
|
# * alive in 3 interpreters (main, interp1, interp2)
|
|
# * module def still in _PyRuntime.imports.extensions
|
|
# * mod init func ran again
|
|
# * m_copy was copied from interp2 (was from interp1)
|
|
# * module's global state was updated, not reset
|
|
|
|
@requires_subinterpreters
|
|
def test_basic_multiple_interpreters_deleted_no_reset(self):
|
|
# without resetting; already loaded in a deleted interpreter
|
|
|
|
# At this point:
|
|
# * alive in 0 interpreters
|
|
# * module def may or may not be loaded already
|
|
# * module def not in _PyRuntime.imports.extensions
|
|
# * mod init func has not run yet (since reset, at least)
|
|
# * m_copy not set (hasn't been loaded yet or already cleared)
|
|
# * module's global state has not been initialized yet
|
|
# (or already cleared)
|
|
|
|
interpid1 = self.add_subinterpreter()
|
|
interpid2 = self.add_subinterpreter()
|
|
|
|
# First, load in the main interpreter but then completely clear it.
|
|
loaded_main = self.load(self.NAME)
|
|
loaded_main.module._clear_globals()
|
|
_testinternalcapi.clear_extension(self.NAME, self.FILE)
|
|
|
|
# At this point:
|
|
# * alive in 0 interpreters
|
|
# * module def loaded already
|
|
# * module def was in _PyRuntime.imports.extensions, but cleared
|
|
# * mod init func ran for the first time (since reset, at least)
|
|
# * m_copy was set, but cleared (was NULL)
|
|
# * module's global state was initialized but cleared
|
|
|
|
# Start with an interpreter that gets destroyed right away.
|
|
base = self.import_in_subinterp(postscript='''
|
|
# Attrs set after loading are not in m_copy.
|
|
mod.spam = 'spam, spam, mash, spam, eggs, and spam'
|
|
''')
|
|
self.check_common(base)
|
|
self.check_fresh(base)
|
|
|
|
# At this point:
|
|
# * alive in 0 interpreters
|
|
# * module def in _PyRuntime.imports.extensions
|
|
# * mod init func ran again
|
|
# * m_copy is NULL (claered when the interpreter was destroyed)
|
|
# * module's global state was initialized, not reset
|
|
|
|
# Use a subinterpreter that sticks around.
|
|
loaded_interp1 = self.import_in_subinterp(interpid1)
|
|
self.check_common(loaded_interp1)
|
|
self.check_semi_fresh(loaded_interp1, loaded_main, base)
|
|
|
|
# At this point:
|
|
# * alive in 1 interpreter (interp1)
|
|
# * module def still in _PyRuntime.imports.extensions
|
|
# * mod init func ran again
|
|
# * m_copy was copied from interp1 (was NULL)
|
|
# * module's global state was updated, not reset
|
|
|
|
# Use a subinterpreter while the previous one is still alive.
|
|
loaded_interp2 = self.import_in_subinterp(interpid2)
|
|
self.check_common(loaded_interp2)
|
|
self.check_copied(loaded_interp2, loaded_interp1)
|
|
|
|
# At this point:
|
|
# * alive in 2 interpreters (interp1, interp2)
|
|
# * module def still in _PyRuntime.imports.extensions
|
|
# * mod init func ran again
|
|
# * m_copy was copied from interp2 (was from interp1)
|
|
# * module's global state was updated, not reset
|
|
|
|
@requires_subinterpreters
|
|
@requires_load_dynamic
|
|
def test_basic_multiple_interpreters_reset_each(self):
|
|
# resetting between each interpreter
|
|
|
|
# At this point:
|
|
# * alive in 0 interpreters
|
|
# * module def may or may not be loaded already
|
|
# * module def not in _PyRuntime.imports.extensions
|
|
# * mod init func has not run yet (since reset, at least)
|
|
# * m_copy not set (hasn't been loaded yet or already cleared)
|
|
# * module's global state has not been initialized yet
|
|
# (or already cleared)
|
|
|
|
interpid1 = self.add_subinterpreter()
|
|
interpid2 = self.add_subinterpreter()
|
|
|
|
# Use an interpreter that gets destroyed right away.
|
|
loaded = self.import_in_subinterp(
|
|
postscript='''
|
|
# Attrs set after loading are not in m_copy.
|
|
mod.spam = 'spam, spam, mash, spam, eggs, and spam'
|
|
''',
|
|
postcleanup=True,
|
|
)
|
|
self.check_common(loaded)
|
|
self.check_fresh(loaded)
|
|
|
|
# At this point:
|
|
# * alive in 0 interpreters
|
|
# * module def in _PyRuntime.imports.extensions
|
|
# * mod init func ran for the first time (since reset, at least)
|
|
# * m_copy is NULL (claered when the interpreter was destroyed)
|
|
# * module's global state was initialized, not reset
|
|
|
|
# Use a subinterpreter that sticks around.
|
|
loaded = self.import_in_subinterp(interpid1, postcleanup=True)
|
|
self.check_common(loaded)
|
|
self.check_fresh(loaded)
|
|
|
|
# At this point:
|
|
# * alive in 1 interpreter (interp1)
|
|
# * module def still in _PyRuntime.imports.extensions
|
|
# * mod init func ran again
|
|
# * m_copy was copied from interp1 (was NULL)
|
|
# * module's global state was initialized, not reset
|
|
|
|
# Use a subinterpreter while the previous one is still alive.
|
|
loaded = self.import_in_subinterp(interpid2, postcleanup=True)
|
|
self.check_common(loaded)
|
|
self.check_fresh(loaded)
|
|
|
|
# At this point:
|
|
# * alive in 2 interpreters (interp2, interp2)
|
|
# * module def still in _PyRuntime.imports.extensions
|
|
# * mod init func ran again
|
|
# * m_copy was copied from interp2 (was from interp1)
|
|
# * module's global state was initialized, not reset
|
|
|
|
|
|
class ReloadTests(unittest.TestCase):
|
|
|
|
"""Very basic tests to make sure that imp.reload() operates just like
|
|
reload()."""
|
|
|
|
def test_source(self):
|
|
# XXX (ncoghlan): It would be nice to use test.import_helper.CleanImport
|
|
# here, but that breaks because the os module registers some
|
|
# handlers in copy_reg on import. Since CleanImport doesn't
|
|
# revert that registration, the module is left in a broken
|
|
# state after reversion. Reinitialising the module contents
|
|
# and just reverting os.environ to its previous state is an OK
|
|
# workaround
|
|
with os_helper.EnvironmentVarGuard():
|
|
import os
|
|
imp.reload(os)
|
|
|
|
def test_extension(self):
|
|
with import_helper.CleanImport('time'):
|
|
import time
|
|
imp.reload(time)
|
|
|
|
def test_builtin(self):
|
|
with import_helper.CleanImport('marshal'):
|
|
import marshal
|
|
imp.reload(marshal)
|
|
|
|
def test_with_deleted_parent(self):
|
|
# see #18681
|
|
from html import parser
|
|
html = sys.modules.pop('html')
|
|
def cleanup():
|
|
sys.modules['html'] = html
|
|
self.addCleanup(cleanup)
|
|
with self.assertRaisesRegex(ImportError, 'html'):
|
|
imp.reload(parser)
|
|
|
|
|
|
class PEP3147Tests(unittest.TestCase):
|
|
"""Tests of PEP 3147."""
|
|
|
|
tag = imp.get_tag()
|
|
|
|
@unittest.skipUnless(sys.implementation.cache_tag is not None,
|
|
'requires sys.implementation.cache_tag not be None')
|
|
def test_cache_from_source(self):
|
|
# Given the path to a .py file, return the path to its PEP 3147
|
|
# defined .pyc file (i.e. under __pycache__).
|
|
path = os.path.join('foo', 'bar', 'baz', 'qux.py')
|
|
expect = os.path.join('foo', 'bar', 'baz', '__pycache__',
|
|
'qux.{}.pyc'.format(self.tag))
|
|
self.assertEqual(imp.cache_from_source(path, True), expect)
|
|
|
|
@unittest.skipUnless(sys.implementation.cache_tag is not None,
|
|
'requires sys.implementation.cache_tag to not be '
|
|
'None')
|
|
def test_source_from_cache(self):
|
|
# Given the path to a PEP 3147 defined .pyc file, return the path to
|
|
# its source. This tests the good path.
|
|
path = os.path.join('foo', 'bar', 'baz', '__pycache__',
|
|
'qux.{}.pyc'.format(self.tag))
|
|
expect = os.path.join('foo', 'bar', 'baz', 'qux.py')
|
|
self.assertEqual(imp.source_from_cache(path), expect)
|
|
|
|
|
|
class NullImporterTests(unittest.TestCase):
|
|
@unittest.skipIf(os_helper.TESTFN_UNENCODABLE is None,
|
|
"Need an undecodeable filename")
|
|
def test_unencodeable(self):
|
|
name = os_helper.TESTFN_UNENCODABLE
|
|
os.mkdir(name)
|
|
try:
|
|
self.assertRaises(ImportError, imp.NullImporter, name)
|
|
finally:
|
|
os.rmdir(name)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|