mirror of https://github.com/python/cpython
907 lines
35 KiB
Python
907 lines
35 KiB
Python
import sys
|
|
import os
|
|
import marshal
|
|
import importlib
|
|
import importlib.util
|
|
import struct
|
|
import time
|
|
import unittest
|
|
import unittest.mock
|
|
import warnings
|
|
|
|
from test import support
|
|
from test.support import import_helper
|
|
from test.support import os_helper
|
|
|
|
from zipfile import ZipFile, ZipInfo, ZIP_STORED, ZIP_DEFLATED
|
|
|
|
import zipimport
|
|
import linecache
|
|
import doctest
|
|
import inspect
|
|
import io
|
|
from traceback import extract_tb, extract_stack, print_tb
|
|
try:
|
|
import zlib
|
|
except ImportError:
|
|
zlib = None
|
|
|
|
test_src = """\
|
|
def get_name():
|
|
return __name__
|
|
def get_file():
|
|
return __file__
|
|
"""
|
|
test_co = compile(test_src, "<???>", "exec")
|
|
raise_src = 'def do_raise(): raise TypeError\n'
|
|
|
|
def make_pyc(co, mtime, size):
|
|
data = marshal.dumps(co)
|
|
pyc = (importlib.util.MAGIC_NUMBER +
|
|
struct.pack("<iLL", 0,
|
|
int(mtime) & 0xFFFF_FFFF, size & 0xFFFF_FFFF) + data)
|
|
return pyc
|
|
|
|
def module_path_to_dotted_name(path):
|
|
return path.replace(os.sep, '.')
|
|
|
|
NOW = time.time()
|
|
test_pyc = make_pyc(test_co, NOW, len(test_src))
|
|
|
|
|
|
TESTMOD = "ziptestmodule"
|
|
TESTPACK = "ziptestpackage"
|
|
TESTPACK2 = "ziptestpackage2"
|
|
TEMP_DIR = os.path.abspath("junk95142")
|
|
TEMP_ZIP = os.path.abspath("junk95142.zip")
|
|
|
|
pyc_file = importlib.util.cache_from_source(TESTMOD + '.py')
|
|
pyc_ext = '.pyc'
|
|
|
|
|
|
class ImportHooksBaseTestCase(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.path = sys.path[:]
|
|
self.meta_path = sys.meta_path[:]
|
|
self.path_hooks = sys.path_hooks[:]
|
|
sys.path_importer_cache.clear()
|
|
self.modules_before = import_helper.modules_setup()
|
|
|
|
def tearDown(self):
|
|
sys.path[:] = self.path
|
|
sys.meta_path[:] = self.meta_path
|
|
sys.path_hooks[:] = self.path_hooks
|
|
sys.path_importer_cache.clear()
|
|
import_helper.modules_cleanup(*self.modules_before)
|
|
|
|
|
|
class UncompressedZipImportTestCase(ImportHooksBaseTestCase):
|
|
|
|
compression = ZIP_STORED
|
|
|
|
def setUp(self):
|
|
# We're reusing the zip archive path, so we must clear the
|
|
# cached directory info and linecache.
|
|
linecache.clearcache()
|
|
zipimport._zip_directory_cache.clear()
|
|
ImportHooksBaseTestCase.setUp(self)
|
|
|
|
def makeTree(self, files, dirName=TEMP_DIR):
|
|
# Create a filesystem based set of modules/packages
|
|
# defined by files under the directory dirName.
|
|
self.addCleanup(os_helper.rmtree, dirName)
|
|
|
|
for name, (mtime, data) in files.items():
|
|
path = os.path.join(dirName, name)
|
|
if path[-1] == os.sep:
|
|
if not os.path.isdir(path):
|
|
os.makedirs(path)
|
|
else:
|
|
dname = os.path.dirname(path)
|
|
if not os.path.isdir(dname):
|
|
os.makedirs(dname)
|
|
with open(path, 'wb') as fp:
|
|
fp.write(data)
|
|
|
|
def makeZip(self, files, zipName=TEMP_ZIP, **kw):
|
|
# Create a zip archive based set of modules/packages
|
|
# defined by files in the zip file zipName. If the
|
|
# key 'stuff' exists in kw it is prepended to the archive.
|
|
self.addCleanup(os_helper.unlink, zipName)
|
|
|
|
with ZipFile(zipName, "w") as z:
|
|
for name, (mtime, data) in files.items():
|
|
zinfo = ZipInfo(name, time.localtime(mtime))
|
|
zinfo.compress_type = self.compression
|
|
z.writestr(zinfo, data)
|
|
comment = kw.get("comment", None)
|
|
if comment is not None:
|
|
z.comment = comment
|
|
|
|
stuff = kw.get("stuff", None)
|
|
if stuff is not None:
|
|
# Prepend 'stuff' to the start of the zipfile
|
|
with open(zipName, "rb") as f:
|
|
data = f.read()
|
|
with open(zipName, "wb") as f:
|
|
f.write(stuff)
|
|
f.write(data)
|
|
|
|
def getZip64Files(self):
|
|
# This is the simplest way to make zipfile generate the zip64 EOCD block
|
|
return {f"f{n}.py": (NOW, test_src) for n in range(65537)}
|
|
|
|
def doTest(self, expected_ext, files, *modules, **kw):
|
|
self.makeZip(files, **kw)
|
|
|
|
sys.path.insert(0, TEMP_ZIP)
|
|
|
|
mod = importlib.import_module(".".join(modules))
|
|
|
|
call = kw.get('call')
|
|
if call is not None:
|
|
call(mod)
|
|
|
|
if expected_ext:
|
|
file = mod.get_file()
|
|
self.assertEqual(file, os.path.join(TEMP_ZIP,
|
|
*modules) + expected_ext)
|
|
|
|
def testAFakeZlib(self):
|
|
#
|
|
# This could cause a stack overflow before: importing zlib.py
|
|
# from a compressed archive would cause zlib to be imported
|
|
# which would find zlib.py in the archive, which would... etc.
|
|
#
|
|
# This test *must* be executed first: it must be the first one
|
|
# to trigger zipimport to import zlib (zipimport caches the
|
|
# zlib.decompress function object, after which the problem being
|
|
# tested here wouldn't be a problem anymore...
|
|
# (Hence the 'A' in the test method name: to make it the first
|
|
# item in a list sorted by name, like
|
|
# unittest.TestLoader.getTestCaseNames() does.)
|
|
#
|
|
# This test fails on platforms on which the zlib module is
|
|
# statically linked, but the problem it tests for can't
|
|
# occur in that case (builtin modules are always found first),
|
|
# so we'll simply skip it then. Bug #765456.
|
|
#
|
|
if "zlib" in sys.builtin_module_names:
|
|
self.skipTest('zlib is a builtin module')
|
|
if "zlib" in sys.modules:
|
|
del sys.modules["zlib"]
|
|
files = {"zlib.py": (NOW, test_src)}
|
|
try:
|
|
self.doTest(".py", files, "zlib")
|
|
except ImportError:
|
|
if self.compression != ZIP_DEFLATED:
|
|
self.fail("expected test to not raise ImportError")
|
|
else:
|
|
if self.compression != ZIP_STORED:
|
|
self.fail("expected test to raise ImportError")
|
|
|
|
def testPy(self):
|
|
files = {TESTMOD + ".py": (NOW, test_src)}
|
|
self.doTest(".py", files, TESTMOD)
|
|
|
|
def testPyc(self):
|
|
files = {TESTMOD + pyc_ext: (NOW, test_pyc)}
|
|
self.doTest(pyc_ext, files, TESTMOD)
|
|
|
|
def testBoth(self):
|
|
files = {TESTMOD + ".py": (NOW, test_src),
|
|
TESTMOD + pyc_ext: (NOW, test_pyc)}
|
|
self.doTest(pyc_ext, files, TESTMOD)
|
|
|
|
def testUncheckedHashBasedPyc(self):
|
|
source = b"state = 'old'"
|
|
source_hash = importlib.util.source_hash(source)
|
|
bytecode = importlib._bootstrap_external._code_to_hash_pyc(
|
|
compile(source, "???", "exec"),
|
|
source_hash,
|
|
False, # unchecked
|
|
)
|
|
files = {TESTMOD + ".py": (NOW, "state = 'new'"),
|
|
TESTMOD + ".pyc": (NOW - 20, bytecode)}
|
|
def check(mod):
|
|
self.assertEqual(mod.state, 'old')
|
|
self.doTest(None, files, TESTMOD, call=check)
|
|
|
|
@unittest.mock.patch('_imp.check_hash_based_pycs', 'always')
|
|
def test_checked_hash_based_change_pyc(self):
|
|
source = b"state = 'old'"
|
|
source_hash = importlib.util.source_hash(source)
|
|
bytecode = importlib._bootstrap_external._code_to_hash_pyc(
|
|
compile(source, "???", "exec"),
|
|
source_hash,
|
|
False,
|
|
)
|
|
files = {TESTMOD + ".py": (NOW, "state = 'new'"),
|
|
TESTMOD + ".pyc": (NOW - 20, bytecode)}
|
|
def check(mod):
|
|
self.assertEqual(mod.state, 'new')
|
|
self.doTest(None, files, TESTMOD, call=check)
|
|
|
|
def testEmptyPy(self):
|
|
files = {TESTMOD + ".py": (NOW, "")}
|
|
self.doTest(None, files, TESTMOD)
|
|
|
|
def testBadMagic(self):
|
|
# make pyc magic word invalid, forcing loading from .py
|
|
badmagic_pyc = bytearray(test_pyc)
|
|
badmagic_pyc[0] ^= 0x04 # flip an arbitrary bit
|
|
files = {TESTMOD + ".py": (NOW, test_src),
|
|
TESTMOD + pyc_ext: (NOW, badmagic_pyc)}
|
|
self.doTest(".py", files, TESTMOD)
|
|
|
|
def testBadMagic2(self):
|
|
# make pyc magic word invalid, causing an ImportError
|
|
badmagic_pyc = bytearray(test_pyc)
|
|
badmagic_pyc[0] ^= 0x04 # flip an arbitrary bit
|
|
files = {TESTMOD + pyc_ext: (NOW, badmagic_pyc)}
|
|
try:
|
|
self.doTest(".py", files, TESTMOD)
|
|
self.fail("This should not be reached")
|
|
except zipimport.ZipImportError as exc:
|
|
self.assertIsInstance(exc.__cause__, ImportError)
|
|
self.assertIn("magic number", exc.__cause__.msg)
|
|
|
|
def testBadMTime(self):
|
|
badtime_pyc = bytearray(test_pyc)
|
|
# flip the second bit -- not the first as that one isn't stored in the
|
|
# .py's mtime in the zip archive.
|
|
badtime_pyc[11] ^= 0x02
|
|
files = {TESTMOD + ".py": (NOW, test_src),
|
|
TESTMOD + pyc_ext: (NOW, badtime_pyc)}
|
|
self.doTest(".py", files, TESTMOD)
|
|
|
|
def test2038MTime(self):
|
|
# Make sure we can handle mtimes larger than what a 32-bit signed number
|
|
# can hold.
|
|
twenty_thirty_eight_pyc = make_pyc(test_co, 2**32 - 1, len(test_src))
|
|
files = {TESTMOD + ".py": (NOW, test_src),
|
|
TESTMOD + pyc_ext: (NOW, twenty_thirty_eight_pyc)}
|
|
self.doTest(".py", files, TESTMOD)
|
|
|
|
def testPackage(self):
|
|
packdir = TESTPACK + os.sep
|
|
files = {packdir + "__init__" + pyc_ext: (NOW, test_pyc),
|
|
packdir + TESTMOD + pyc_ext: (NOW, test_pyc)}
|
|
self.doTest(pyc_ext, files, TESTPACK, TESTMOD)
|
|
|
|
def testSubPackage(self):
|
|
# Test that subpackages function when loaded from zip
|
|
# archives.
|
|
packdir = TESTPACK + os.sep
|
|
packdir2 = packdir + TESTPACK2 + os.sep
|
|
files = {packdir + "__init__" + pyc_ext: (NOW, test_pyc),
|
|
packdir2 + "__init__" + pyc_ext: (NOW, test_pyc),
|
|
packdir2 + TESTMOD + pyc_ext: (NOW, test_pyc)}
|
|
self.doTest(pyc_ext, files, TESTPACK, TESTPACK2, TESTMOD)
|
|
|
|
def testSubNamespacePackage(self):
|
|
# Test that implicit namespace subpackages function
|
|
# when loaded from zip archives.
|
|
packdir = TESTPACK + os.sep
|
|
packdir2 = packdir + TESTPACK2 + os.sep
|
|
# The first two files are just directory entries (so have no data).
|
|
files = {packdir: (NOW, ""),
|
|
packdir2: (NOW, ""),
|
|
packdir2 + TESTMOD + pyc_ext: (NOW, test_pyc)}
|
|
self.doTest(pyc_ext, files, TESTPACK, TESTPACK2, TESTMOD)
|
|
|
|
def testMixedNamespacePackage(self):
|
|
# Test implicit namespace packages spread between a
|
|
# real filesystem and a zip archive.
|
|
packdir = TESTPACK + os.sep
|
|
packdir2 = packdir + TESTPACK2 + os.sep
|
|
packdir3 = packdir2 + TESTPACK + '3' + os.sep
|
|
files1 = {packdir: (NOW, ""),
|
|
packdir + TESTMOD + pyc_ext: (NOW, test_pyc),
|
|
packdir2: (NOW, ""),
|
|
packdir3: (NOW, ""),
|
|
packdir3 + TESTMOD + pyc_ext: (NOW, test_pyc),
|
|
packdir2 + TESTMOD + '3' + pyc_ext: (NOW, test_pyc),
|
|
packdir2 + TESTMOD + pyc_ext: (NOW, test_pyc)}
|
|
files2 = {packdir: (NOW, ""),
|
|
packdir + TESTMOD + '2' + pyc_ext: (NOW, test_pyc),
|
|
packdir2: (NOW, ""),
|
|
packdir2 + TESTMOD + '2' + pyc_ext: (NOW, test_pyc),
|
|
packdir2 + TESTMOD + pyc_ext: (NOW, test_pyc)}
|
|
|
|
zip1 = os.path.abspath("path1.zip")
|
|
self.makeZip(files1, zip1)
|
|
|
|
zip2 = TEMP_DIR
|
|
self.makeTree(files2, zip2)
|
|
|
|
# zip2 should override zip1.
|
|
sys.path.insert(0, zip1)
|
|
sys.path.insert(0, zip2)
|
|
|
|
mod = importlib.import_module(TESTPACK)
|
|
|
|
# if TESTPACK is functioning as a namespace pkg then
|
|
# there should be two entries in the __path__.
|
|
# First should be path2 and second path1.
|
|
self.assertEqual(2, len(mod.__path__))
|
|
p1, p2 = mod.__path__
|
|
self.assertEqual(os.path.basename(TEMP_DIR), p1.split(os.sep)[-2])
|
|
self.assertEqual("path1.zip", p2.split(os.sep)[-2])
|
|
|
|
# packdir3 should import as a namespace package.
|
|
# Its __path__ is an iterable of 1 element from zip1.
|
|
mod = importlib.import_module(packdir3.replace(os.sep, '.')[:-1])
|
|
self.assertEqual(1, len(mod.__path__))
|
|
mpath = list(mod.__path__)[0].split('path1.zip' + os.sep)[1]
|
|
self.assertEqual(packdir3[:-1], mpath)
|
|
|
|
# TESTPACK/TESTMOD only exists in path1.
|
|
mod = importlib.import_module('.'.join((TESTPACK, TESTMOD)))
|
|
self.assertEqual("path1.zip", mod.__file__.split(os.sep)[-3])
|
|
|
|
# And TESTPACK/(TESTMOD + '2') only exists in path2.
|
|
mod = importlib.import_module('.'.join((TESTPACK, TESTMOD + '2')))
|
|
self.assertEqual(os.path.basename(TEMP_DIR),
|
|
mod.__file__.split(os.sep)[-3])
|
|
|
|
# One level deeper...
|
|
subpkg = '.'.join((TESTPACK, TESTPACK2))
|
|
mod = importlib.import_module(subpkg)
|
|
self.assertEqual(2, len(mod.__path__))
|
|
p1, p2 = mod.__path__
|
|
self.assertEqual(os.path.basename(TEMP_DIR), p1.split(os.sep)[-3])
|
|
self.assertEqual("path1.zip", p2.split(os.sep)[-3])
|
|
|
|
# subpkg.TESTMOD exists in both zips should load from zip2.
|
|
mod = importlib.import_module('.'.join((subpkg, TESTMOD)))
|
|
self.assertEqual(os.path.basename(TEMP_DIR),
|
|
mod.__file__.split(os.sep)[-4])
|
|
|
|
# subpkg.TESTMOD + '2' only exists in zip2.
|
|
mod = importlib.import_module('.'.join((subpkg, TESTMOD + '2')))
|
|
self.assertEqual(os.path.basename(TEMP_DIR),
|
|
mod.__file__.split(os.sep)[-4])
|
|
|
|
# Finally subpkg.TESTMOD + '3' only exists in zip1.
|
|
mod = importlib.import_module('.'.join((subpkg, TESTMOD + '3')))
|
|
self.assertEqual('path1.zip', mod.__file__.split(os.sep)[-4])
|
|
|
|
def testNamespacePackage(self):
|
|
# Test implicit namespace packages spread between multiple zip
|
|
# archives.
|
|
packdir = TESTPACK + os.sep
|
|
packdir2 = packdir + TESTPACK2 + os.sep
|
|
packdir3 = packdir2 + TESTPACK + '3' + os.sep
|
|
files1 = {packdir: (NOW, ""),
|
|
packdir + TESTMOD + pyc_ext: (NOW, test_pyc),
|
|
packdir2: (NOW, ""),
|
|
packdir3: (NOW, ""),
|
|
packdir3 + TESTMOD + pyc_ext: (NOW, test_pyc),
|
|
packdir2 + TESTMOD + '3' + pyc_ext: (NOW, test_pyc),
|
|
packdir2 + TESTMOD + pyc_ext: (NOW, test_pyc)}
|
|
zip1 = os.path.abspath("path1.zip")
|
|
self.makeZip(files1, zip1)
|
|
|
|
files2 = {packdir: (NOW, ""),
|
|
packdir + TESTMOD + '2' + pyc_ext: (NOW, test_pyc),
|
|
packdir2: (NOW, ""),
|
|
packdir2 + TESTMOD + '2' + pyc_ext: (NOW, test_pyc),
|
|
packdir2 + TESTMOD + pyc_ext: (NOW, test_pyc)}
|
|
zip2 = os.path.abspath("path2.zip")
|
|
self.makeZip(files2, zip2)
|
|
|
|
# zip2 should override zip1.
|
|
sys.path.insert(0, zip1)
|
|
sys.path.insert(0, zip2)
|
|
|
|
mod = importlib.import_module(TESTPACK)
|
|
|
|
# if TESTPACK is functioning as a namespace pkg then
|
|
# there should be two entries in the __path__.
|
|
# First should be path2 and second path1.
|
|
self.assertEqual(2, len(mod.__path__))
|
|
p1, p2 = mod.__path__
|
|
self.assertEqual("path2.zip", p1.split(os.sep)[-2])
|
|
self.assertEqual("path1.zip", p2.split(os.sep)[-2])
|
|
|
|
# packdir3 should import as a namespace package.
|
|
# Tts __path__ is an iterable of 1 element from zip1.
|
|
mod = importlib.import_module(packdir3.replace(os.sep, '.')[:-1])
|
|
self.assertEqual(1, len(mod.__path__))
|
|
mpath = list(mod.__path__)[0].split('path1.zip' + os.sep)[1]
|
|
self.assertEqual(packdir3[:-1], mpath)
|
|
|
|
# TESTPACK/TESTMOD only exists in path1.
|
|
mod = importlib.import_module('.'.join((TESTPACK, TESTMOD)))
|
|
self.assertEqual("path1.zip", mod.__file__.split(os.sep)[-3])
|
|
|
|
# And TESTPACK/(TESTMOD + '2') only exists in path2.
|
|
mod = importlib.import_module('.'.join((TESTPACK, TESTMOD + '2')))
|
|
self.assertEqual("path2.zip", mod.__file__.split(os.sep)[-3])
|
|
|
|
# One level deeper...
|
|
subpkg = '.'.join((TESTPACK, TESTPACK2))
|
|
mod = importlib.import_module(subpkg)
|
|
self.assertEqual(2, len(mod.__path__))
|
|
p1, p2 = mod.__path__
|
|
self.assertEqual("path2.zip", p1.split(os.sep)[-3])
|
|
self.assertEqual("path1.zip", p2.split(os.sep)[-3])
|
|
|
|
# subpkg.TESTMOD exists in both zips should load from zip2.
|
|
mod = importlib.import_module('.'.join((subpkg, TESTMOD)))
|
|
self.assertEqual('path2.zip', mod.__file__.split(os.sep)[-4])
|
|
|
|
# subpkg.TESTMOD + '2' only exists in zip2.
|
|
mod = importlib.import_module('.'.join((subpkg, TESTMOD + '2')))
|
|
self.assertEqual('path2.zip', mod.__file__.split(os.sep)[-4])
|
|
|
|
# Finally subpkg.TESTMOD + '3' only exists in zip1.
|
|
mod = importlib.import_module('.'.join((subpkg, TESTMOD + '3')))
|
|
self.assertEqual('path1.zip', mod.__file__.split(os.sep)[-4])
|
|
|
|
def testZipImporterMethods(self):
|
|
packdir = TESTPACK + os.sep
|
|
packdir2 = packdir + TESTPACK2 + os.sep
|
|
files = {packdir + "__init__" + pyc_ext: (NOW, test_pyc),
|
|
packdir2 + "__init__" + pyc_ext: (NOW, test_pyc),
|
|
packdir2 + TESTMOD + pyc_ext: (NOW, test_pyc),
|
|
"spam" + pyc_ext: (NOW, test_pyc)}
|
|
|
|
self.addCleanup(os_helper.unlink, TEMP_ZIP)
|
|
with ZipFile(TEMP_ZIP, "w") as z:
|
|
for name, (mtime, data) in files.items():
|
|
zinfo = ZipInfo(name, time.localtime(mtime))
|
|
zinfo.compress_type = self.compression
|
|
zinfo.comment = b"spam"
|
|
z.writestr(zinfo, data)
|
|
|
|
zi = zipimport.zipimporter(TEMP_ZIP)
|
|
self.assertEqual(zi.archive, TEMP_ZIP)
|
|
self.assertTrue(zi.is_package(TESTPACK))
|
|
|
|
# PEP 302
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
|
|
mod = zi.load_module(TESTPACK)
|
|
self.assertEqual(zi.get_filename(TESTPACK), mod.__file__)
|
|
|
|
# PEP 451
|
|
spec = zi.find_spec('spam')
|
|
self.assertIsNotNone(spec)
|
|
self.assertIsInstance(spec.loader, zipimport.zipimporter)
|
|
self.assertFalse(spec.loader.is_package('spam'))
|
|
exec_mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(exec_mod)
|
|
self.assertEqual(spec.loader.get_filename('spam'), exec_mod.__file__)
|
|
|
|
spec = zi.find_spec(TESTPACK)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
self.assertEqual(zi.get_filename(TESTPACK), mod.__file__)
|
|
|
|
existing_pack_path = importlib.import_module(TESTPACK).__path__[0]
|
|
expected_path_path = os.path.join(TEMP_ZIP, TESTPACK)
|
|
self.assertEqual(existing_pack_path, expected_path_path)
|
|
|
|
self.assertFalse(zi.is_package(packdir + '__init__'))
|
|
self.assertTrue(zi.is_package(packdir + TESTPACK2))
|
|
self.assertFalse(zi.is_package(packdir2 + TESTMOD))
|
|
|
|
mod_path = packdir2 + TESTMOD
|
|
mod_name = module_path_to_dotted_name(mod_path)
|
|
mod = importlib.import_module(mod_name)
|
|
self.assertTrue(mod_name in sys.modules)
|
|
self.assertIsNone(zi.get_source(TESTPACK))
|
|
self.assertIsNone(zi.get_source(mod_path))
|
|
self.assertEqual(zi.get_filename(mod_path), mod.__file__)
|
|
# To pass in the module name instead of the path, we must use the
|
|
# right importer
|
|
loader = mod.__spec__.loader
|
|
self.assertIsNone(loader.get_source(mod_name))
|
|
self.assertEqual(loader.get_filename(mod_name), mod.__file__)
|
|
|
|
# test prefix and archivepath members
|
|
zi2 = zipimport.zipimporter(TEMP_ZIP + os.sep + TESTPACK)
|
|
self.assertEqual(zi2.archive, TEMP_ZIP)
|
|
self.assertEqual(zi2.prefix, TESTPACK + os.sep)
|
|
|
|
def testInvalidateCaches(self):
|
|
packdir = TESTPACK + os.sep
|
|
packdir2 = packdir + TESTPACK2 + os.sep
|
|
files = {packdir + "__init__" + pyc_ext: (NOW, test_pyc),
|
|
packdir2 + "__init__" + pyc_ext: (NOW, test_pyc),
|
|
packdir2 + TESTMOD + pyc_ext: (NOW, test_pyc),
|
|
"spam" + pyc_ext: (NOW, test_pyc)}
|
|
self.addCleanup(os_helper.unlink, TEMP_ZIP)
|
|
with ZipFile(TEMP_ZIP, "w") as z:
|
|
for name, (mtime, data) in files.items():
|
|
zinfo = ZipInfo(name, time.localtime(mtime))
|
|
zinfo.compress_type = self.compression
|
|
zinfo.comment = b"spam"
|
|
z.writestr(zinfo, data)
|
|
|
|
zi = zipimport.zipimporter(TEMP_ZIP)
|
|
self.assertEqual(zi._get_files().keys(), files.keys())
|
|
# Check that the file information remains accurate after reloading
|
|
zi.invalidate_caches()
|
|
self.assertEqual(zi._get_files().keys(), files.keys())
|
|
# Add a new file to the ZIP archive
|
|
newfile = {"spam2" + pyc_ext: (NOW, test_pyc)}
|
|
files.update(newfile)
|
|
with ZipFile(TEMP_ZIP, "a") as z:
|
|
for name, (mtime, data) in newfile.items():
|
|
zinfo = ZipInfo(name, time.localtime(mtime))
|
|
zinfo.compress_type = self.compression
|
|
zinfo.comment = b"spam"
|
|
z.writestr(zinfo, data)
|
|
# Check that we can detect the new file after invalidating the cache
|
|
zi.invalidate_caches()
|
|
self.assertEqual(zi._get_files().keys(), files.keys())
|
|
spec = zi.find_spec('spam2')
|
|
self.assertIsNotNone(spec)
|
|
self.assertIsInstance(spec.loader, zipimport.zipimporter)
|
|
# Check that the cached data is removed if the file is deleted
|
|
os.remove(TEMP_ZIP)
|
|
zi.invalidate_caches()
|
|
self.assertFalse(zi._get_files())
|
|
self.assertIsNone(zipimport._zip_directory_cache.get(zi.archive))
|
|
self.assertIsNone(zi.find_spec("name_does_not_matter"))
|
|
|
|
def testInvalidateCachesWithMultipleZipimports(self):
|
|
packdir = TESTPACK + os.sep
|
|
packdir2 = packdir + TESTPACK2 + os.sep
|
|
files = {packdir + "__init__" + pyc_ext: (NOW, test_pyc),
|
|
packdir2 + "__init__" + pyc_ext: (NOW, test_pyc),
|
|
packdir2 + TESTMOD + pyc_ext: (NOW, test_pyc),
|
|
"spam" + pyc_ext: (NOW, test_pyc)}
|
|
self.addCleanup(os_helper.unlink, TEMP_ZIP)
|
|
with ZipFile(TEMP_ZIP, "w") as z:
|
|
for name, (mtime, data) in files.items():
|
|
zinfo = ZipInfo(name, time.localtime(mtime))
|
|
zinfo.compress_type = self.compression
|
|
zinfo.comment = b"spam"
|
|
z.writestr(zinfo, data)
|
|
|
|
zi = zipimport.zipimporter(TEMP_ZIP)
|
|
self.assertEqual(zi._get_files().keys(), files.keys())
|
|
# Zipimporter for the same path.
|
|
zi2 = zipimport.zipimporter(TEMP_ZIP)
|
|
self.assertEqual(zi2._get_files().keys(), files.keys())
|
|
# Add a new file to the ZIP archive to make the cache wrong.
|
|
newfile = {"spam2" + pyc_ext: (NOW, test_pyc)}
|
|
files.update(newfile)
|
|
with ZipFile(TEMP_ZIP, "a") as z:
|
|
for name, (mtime, data) in newfile.items():
|
|
zinfo = ZipInfo(name, time.localtime(mtime))
|
|
zinfo.compress_type = self.compression
|
|
zinfo.comment = b"spam"
|
|
z.writestr(zinfo, data)
|
|
# Invalidate the cache of the first zipimporter.
|
|
zi.invalidate_caches()
|
|
# Check that the second zipimporter detects the new file and isn't using a stale cache.
|
|
self.assertEqual(zi2._get_files().keys(), files.keys())
|
|
spec = zi2.find_spec('spam2')
|
|
self.assertIsNotNone(spec)
|
|
self.assertIsInstance(spec.loader, zipimport.zipimporter)
|
|
|
|
def testZipImporterMethodsInSubDirectory(self):
|
|
packdir = TESTPACK + os.sep
|
|
packdir2 = packdir + TESTPACK2 + os.sep
|
|
files = {packdir2 + "__init__" + pyc_ext: (NOW, test_pyc),
|
|
packdir2 + TESTMOD + pyc_ext: (NOW, test_pyc)}
|
|
|
|
self.addCleanup(os_helper.unlink, TEMP_ZIP)
|
|
with ZipFile(TEMP_ZIP, "w") as z:
|
|
for name, (mtime, data) in files.items():
|
|
zinfo = ZipInfo(name, time.localtime(mtime))
|
|
zinfo.compress_type = self.compression
|
|
zinfo.comment = b"eggs"
|
|
z.writestr(zinfo, data)
|
|
|
|
zi = zipimport.zipimporter(TEMP_ZIP + os.sep + packdir)
|
|
self.assertEqual(zi.archive, TEMP_ZIP)
|
|
self.assertEqual(zi.prefix, packdir)
|
|
self.assertTrue(zi.is_package(TESTPACK2))
|
|
# PEP 302
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
mod = zi.load_module(TESTPACK2)
|
|
self.assertEqual(zi.get_filename(TESTPACK2), mod.__file__)
|
|
# PEP 451
|
|
spec = zi.find_spec(TESTPACK2)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
self.assertEqual(spec.loader.get_filename(TESTPACK2), mod.__file__)
|
|
|
|
self.assertFalse(zi.is_package(TESTPACK2 + os.sep + '__init__'))
|
|
self.assertFalse(zi.is_package(TESTPACK2 + os.sep + TESTMOD))
|
|
|
|
pkg_path = TEMP_ZIP + os.sep + packdir + TESTPACK2
|
|
zi2 = zipimport.zipimporter(pkg_path)
|
|
|
|
# PEP 451
|
|
spec = zi2.find_spec(TESTMOD)
|
|
self.assertIsNotNone(spec)
|
|
self.assertIsInstance(spec.loader, zipimport.zipimporter)
|
|
self.assertFalse(spec.loader.is_package(TESTMOD))
|
|
load_mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(load_mod)
|
|
self.assertEqual(
|
|
spec.loader.get_filename(TESTMOD), load_mod.__file__)
|
|
|
|
mod_path = TESTPACK2 + os.sep + TESTMOD
|
|
mod_name = module_path_to_dotted_name(mod_path)
|
|
mod = importlib.import_module(mod_name)
|
|
self.assertTrue(mod_name in sys.modules)
|
|
self.assertIsNone(zi.get_source(TESTPACK2))
|
|
self.assertIsNone(zi.get_source(mod_path))
|
|
self.assertEqual(zi.get_filename(mod_path), mod.__file__)
|
|
# To pass in the module name instead of the path, we must use the
|
|
# right importer.
|
|
loader = mod.__loader__
|
|
self.assertIsNone(loader.get_source(mod_name))
|
|
self.assertEqual(loader.get_filename(mod_name), mod.__file__)
|
|
|
|
def testGetData(self):
|
|
self.addCleanup(os_helper.unlink, TEMP_ZIP)
|
|
with ZipFile(TEMP_ZIP, "w") as z:
|
|
z.compression = self.compression
|
|
name = "testdata.dat"
|
|
data = bytes(x for x in range(256))
|
|
z.writestr(name, data)
|
|
|
|
zi = zipimport.zipimporter(TEMP_ZIP)
|
|
self.assertEqual(data, zi.get_data(name))
|
|
self.assertIn('zipimporter object', repr(zi))
|
|
|
|
def testImporterAttr(self):
|
|
src = """if 1: # indent hack
|
|
def get_file():
|
|
return __file__
|
|
if __loader__.get_data("some.data") != b"some data":
|
|
raise AssertionError("bad data")\n"""
|
|
pyc = make_pyc(compile(src, "<???>", "exec"), NOW, len(src))
|
|
files = {TESTMOD + pyc_ext: (NOW, pyc),
|
|
"some.data": (NOW, "some data")}
|
|
self.doTest(pyc_ext, files, TESTMOD)
|
|
|
|
def testDefaultOptimizationLevel(self):
|
|
# zipimport should use the default optimization level (#28131)
|
|
src = """if 1: # indent hack
|
|
def test(val):
|
|
assert(val)
|
|
return val\n"""
|
|
files = {TESTMOD + '.py': (NOW, src)}
|
|
self.makeZip(files)
|
|
sys.path.insert(0, TEMP_ZIP)
|
|
mod = importlib.import_module(TESTMOD)
|
|
self.assertEqual(mod.test(1), 1)
|
|
if __debug__:
|
|
self.assertRaises(AssertionError, mod.test, False)
|
|
else:
|
|
self.assertEqual(mod.test(0), 0)
|
|
|
|
def testImport_WithStuff(self):
|
|
# try importing from a zipfile which contains additional
|
|
# stuff at the beginning of the file
|
|
files = {TESTMOD + ".py": (NOW, test_src)}
|
|
self.doTest(".py", files, TESTMOD,
|
|
stuff=b"Some Stuff"*31)
|
|
|
|
def assertModuleSource(self, module):
|
|
self.assertEqual(inspect.getsource(module), test_src)
|
|
|
|
def testGetSource(self):
|
|
files = {TESTMOD + ".py": (NOW, test_src)}
|
|
self.doTest(".py", files, TESTMOD, call=self.assertModuleSource)
|
|
|
|
def testGetCompiledSource(self):
|
|
pyc = make_pyc(compile(test_src, "<???>", "exec"), NOW, len(test_src))
|
|
files = {TESTMOD + ".py": (NOW, test_src),
|
|
TESTMOD + pyc_ext: (NOW, pyc)}
|
|
self.doTest(pyc_ext, files, TESTMOD, call=self.assertModuleSource)
|
|
|
|
def runDoctest(self, callback):
|
|
files = {TESTMOD + ".py": (NOW, test_src),
|
|
"xyz.txt": (NOW, ">>> log.append(True)\n")}
|
|
self.doTest(".py", files, TESTMOD, call=callback)
|
|
|
|
def doDoctestFile(self, module):
|
|
log = []
|
|
old_master, doctest.master = doctest.master, None
|
|
try:
|
|
doctest.testfile(
|
|
'xyz.txt', package=module, module_relative=True,
|
|
globs=locals()
|
|
)
|
|
finally:
|
|
doctest.master = old_master
|
|
self.assertEqual(log,[True])
|
|
|
|
def testDoctestFile(self):
|
|
self.runDoctest(self.doDoctestFile)
|
|
|
|
def doDoctestSuite(self, module):
|
|
log = []
|
|
doctest.DocFileTest(
|
|
'xyz.txt', package=module, module_relative=True,
|
|
globs=locals()
|
|
).run()
|
|
self.assertEqual(log,[True])
|
|
|
|
def testDoctestSuite(self):
|
|
self.runDoctest(self.doDoctestSuite)
|
|
|
|
def doTraceback(self, module):
|
|
try:
|
|
module.do_raise()
|
|
except Exception as e:
|
|
tb = e.__traceback__.tb_next
|
|
|
|
f,lno,n,line = extract_tb(tb, 1)[0]
|
|
self.assertEqual(line, raise_src.strip())
|
|
|
|
f,lno,n,line = extract_stack(tb.tb_frame, 1)[0]
|
|
self.assertEqual(line, raise_src.strip())
|
|
|
|
s = io.StringIO()
|
|
print_tb(tb, 1, s)
|
|
self.assertTrue(s.getvalue().endswith(
|
|
' def do_raise(): raise TypeError\n'
|
|
'' if support.has_no_debug_ranges() else
|
|
' ^^^^^^^^^^^^^^^\n'
|
|
))
|
|
else:
|
|
raise AssertionError("This ought to be impossible")
|
|
|
|
def testTraceback(self):
|
|
files = {TESTMOD + ".py": (NOW, raise_src)}
|
|
self.doTest(None, files, TESTMOD, call=self.doTraceback)
|
|
|
|
@unittest.skipIf(os_helper.TESTFN_UNENCODABLE is None,
|
|
"need an unencodable filename")
|
|
def testUnencodable(self):
|
|
filename = os_helper.TESTFN_UNENCODABLE + ".zip"
|
|
self.addCleanup(os_helper.unlink, filename)
|
|
with ZipFile(filename, "w") as z:
|
|
zinfo = ZipInfo(TESTMOD + ".py", time.localtime(NOW))
|
|
zinfo.compress_type = self.compression
|
|
z.writestr(zinfo, test_src)
|
|
spec = zipimport.zipimporter(filename).find_spec(TESTMOD)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
|
|
def testBytesPath(self):
|
|
filename = os_helper.TESTFN + ".zip"
|
|
self.addCleanup(os_helper.unlink, filename)
|
|
with ZipFile(filename, "w") as z:
|
|
zinfo = ZipInfo(TESTMOD + ".py", time.localtime(NOW))
|
|
zinfo.compress_type = self.compression
|
|
z.writestr(zinfo, test_src)
|
|
|
|
zipimport.zipimporter(filename)
|
|
with self.assertRaises(TypeError):
|
|
zipimport.zipimporter(os.fsencode(filename))
|
|
with self.assertRaises(TypeError):
|
|
zipimport.zipimporter(bytearray(os.fsencode(filename)))
|
|
with self.assertRaises(TypeError):
|
|
zipimport.zipimporter(memoryview(os.fsencode(filename)))
|
|
|
|
def testComment(self):
|
|
files = {TESTMOD + ".py": (NOW, test_src)}
|
|
self.doTest(".py", files, TESTMOD, comment=b"comment")
|
|
|
|
def testBeginningCruftAndComment(self):
|
|
files = {TESTMOD + ".py": (NOW, test_src)}
|
|
self.doTest(".py", files, TESTMOD, stuff=b"cruft" * 64, comment=b"hi")
|
|
|
|
def testLargestPossibleComment(self):
|
|
files = {TESTMOD + ".py": (NOW, test_src)}
|
|
self.doTest(".py", files, TESTMOD, comment=b"c" * ((1 << 16) - 1))
|
|
|
|
def testZip64(self):
|
|
files = self.getZip64Files()
|
|
self.doTest(".py", files, "f6")
|
|
|
|
def testZip64CruftAndComment(self):
|
|
files = self.getZip64Files()
|
|
self.doTest(".py", files, "f65536", comment=b"c" * ((1 << 16) - 1))
|
|
|
|
|
|
@support.requires_zlib()
|
|
class CompressedZipImportTestCase(UncompressedZipImportTestCase):
|
|
compression = ZIP_DEFLATED
|
|
|
|
|
|
class BadFileZipImportTestCase(unittest.TestCase):
|
|
def assertZipFailure(self, filename):
|
|
self.assertRaises(zipimport.ZipImportError,
|
|
zipimport.zipimporter, filename)
|
|
|
|
def testNoFile(self):
|
|
self.assertZipFailure('AdfjdkFJKDFJjdklfjs')
|
|
|
|
def testEmptyFilename(self):
|
|
self.assertZipFailure('')
|
|
|
|
def testBadArgs(self):
|
|
self.assertRaises(TypeError, zipimport.zipimporter, None)
|
|
self.assertRaises(TypeError, zipimport.zipimporter, TESTMOD, kwd=None)
|
|
self.assertRaises(TypeError, zipimport.zipimporter,
|
|
list(os.fsencode(TESTMOD)))
|
|
|
|
def testFilenameTooLong(self):
|
|
self.assertZipFailure('A' * 33000)
|
|
|
|
def testEmptyFile(self):
|
|
os_helper.unlink(TESTMOD)
|
|
os_helper.create_empty_file(TESTMOD)
|
|
self.assertZipFailure(TESTMOD)
|
|
|
|
@unittest.skipIf(support.is_wasi, "mode 000 not supported.")
|
|
def testFileUnreadable(self):
|
|
os_helper.unlink(TESTMOD)
|
|
fd = os.open(TESTMOD, os.O_CREAT, 000)
|
|
try:
|
|
os.close(fd)
|
|
|
|
with self.assertRaises(zipimport.ZipImportError) as cm:
|
|
zipimport.zipimporter(TESTMOD)
|
|
finally:
|
|
# If we leave "the read-only bit" set on Windows, nothing can
|
|
# delete TESTMOD, and later tests suffer bogus failures.
|
|
os.chmod(TESTMOD, 0o666)
|
|
os_helper.unlink(TESTMOD)
|
|
|
|
def testNotZipFile(self):
|
|
os_helper.unlink(TESTMOD)
|
|
fp = open(TESTMOD, 'w+')
|
|
fp.write('a' * 22)
|
|
fp.close()
|
|
self.assertZipFailure(TESTMOD)
|
|
|
|
# XXX: disabled until this works on Big-endian machines
|
|
def _testBogusZipFile(self):
|
|
os_helper.unlink(TESTMOD)
|
|
fp = open(TESTMOD, 'w+')
|
|
fp.write(struct.pack('=I', 0x06054B50))
|
|
fp.write('a' * 18)
|
|
fp.close()
|
|
z = zipimport.zipimporter(TESTMOD)
|
|
|
|
try:
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
self.assertRaises(TypeError, z.load_module, None)
|
|
self.assertRaises(TypeError, z.find_module, None)
|
|
self.assertRaises(TypeError, z.find_spec, None)
|
|
self.assertRaises(TypeError, z.exec_module, None)
|
|
self.assertRaises(TypeError, z.is_package, None)
|
|
self.assertRaises(TypeError, z.get_code, None)
|
|
self.assertRaises(TypeError, z.get_data, None)
|
|
self.assertRaises(TypeError, z.get_source, None)
|
|
|
|
error = zipimport.ZipImportError
|
|
self.assertIsNone(z.find_spec('abc'))
|
|
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
self.assertRaises(error, z.load_module, 'abc')
|
|
self.assertRaises(error, z.get_code, 'abc')
|
|
self.assertRaises(OSError, z.get_data, 'abc')
|
|
self.assertRaises(error, z.get_source, 'abc')
|
|
self.assertRaises(error, z.is_package, 'abc')
|
|
finally:
|
|
zipimport._zip_directory_cache.clear()
|
|
|
|
|
|
def tearDownModule():
|
|
os_helper.unlink(TESTMOD)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|