mirror of
https://github.com/python/cpython.git
synced 2025-07-24 03:35:53 +00:00
bb0cf8fd60
This is related to fixing the refleaks introduced by commit 096d009. I haven't been able to find the leak yet, but these changes are a consequence of that effort. This includes some cleanup, some tweaks to the existing tests, and a bunch of new test cases. The only change here that might have impact outside the tests in question is in imp.py, where I update imp.load_dynamic() to use spec_from_file_location() instead of creating a ModuleSpec directly.
Also note that I've updated the tests to only skip if we're checking for refleaks (regrtest's --huntrleaks), whereas in gh-101969 I had skipped the tests entirely. The tests will be useful for some upcoming work and I'd rather the refleaks not hold that up. (It isn't clear how quickly we'll be able to fix the leaking code, though it will certainly be done in the short term.)
https://github.com/python/cpython/issues/102251
1364 lines
50 KiB
Python
1364 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()
|