mirror of
https://github.com/python/cpython.git
synced 2025-08-03 08:34:29 +00:00
PEP 302 + zipimport:
- new import hooks in import.c, exposed in the sys module - new module called 'zipimport' - various changes to allow bootstrapping from zip files I hope I didn't break the Windows build (or anything else for that matter), but then again, it's been sitting on sf long enough... Regarding the latest discussions on python-dev: zipimport sets pkg.__path__ as specified in PEP 273, and likewise, sys.path item such as /path/to/Archive.zip/subdir/ are supported again.
This commit is contained in:
parent
60087fb450
commit
52e14d640b
13 changed files with 1887 additions and 35 deletions
13
Lib/site.py
13
Lib/site.py
|
@ -73,16 +73,11 @@ del m
|
|||
# only absolute pathnames, even if we're running from the build directory.
|
||||
L = []
|
||||
_dirs_in_sys_path = {}
|
||||
dir = dircase = None # sys.path may be empty at this point
|
||||
for dir in sys.path:
|
||||
# Filter out paths that don't exist, but leave in the empty string
|
||||
# since it's a special case. We also need to special-case the Mac,
|
||||
# as file names are allowed on sys.path there.
|
||||
if sys.platform != 'mac':
|
||||
if dir and not os.path.isdir(dir):
|
||||
continue
|
||||
else:
|
||||
if dir and not os.path.exists(dir):
|
||||
continue
|
||||
# Filter out duplicate paths (on case-insensitive file systems also
|
||||
# if they only differ in case); turn relative paths into absolute
|
||||
# paths.
|
||||
dir, dircase = makepath(dir)
|
||||
if not dircase in _dirs_in_sys_path:
|
||||
L.append(dir)
|
||||
|
|
204
Lib/test/test_importhooks.py
Normal file
204
Lib/test/test_importhooks.py
Normal file
|
@ -0,0 +1,204 @@
|
|||
import sys
|
||||
import imp
|
||||
import os
|
||||
import unittest
|
||||
from test import test_support
|
||||
|
||||
|
||||
test_src = """\
|
||||
def get_name():
|
||||
return __name__
|
||||
def get_file():
|
||||
return __file__
|
||||
"""
|
||||
|
||||
test_co = compile(test_src, "<???>", "exec")
|
||||
test_path = "!!!_test_!!!"
|
||||
|
||||
|
||||
class ImportTracker:
|
||||
"""Importer that only tracks attempted imports."""
|
||||
def __init__(self):
|
||||
self.imports = []
|
||||
def find_module(self, fullname, path=None):
|
||||
self.imports.append(fullname)
|
||||
return None
|
||||
|
||||
|
||||
class TestImporter:
|
||||
|
||||
modules = {
|
||||
"hooktestmodule": (False, test_co),
|
||||
"hooktestpackage": (True, test_co),
|
||||
"hooktestpackage.sub": (True, test_co),
|
||||
"hooktestpackage.sub.subber": (False, test_co),
|
||||
}
|
||||
|
||||
def __init__(self, path=test_path):
|
||||
if path != test_path:
|
||||
# if out class is on sys.path_hooks, we must raise
|
||||
# ImportError for any path item that we can't handle.
|
||||
raise ImportError
|
||||
self.path = path
|
||||
|
||||
def _get__path__(self):
|
||||
raise NotImplementedError
|
||||
|
||||
def find_module(self, fullname, path=None):
|
||||
if fullname in self.modules:
|
||||
return self
|
||||
else:
|
||||
return None
|
||||
|
||||
def load_module(self, fullname):
|
||||
ispkg, code = self.modules[fullname]
|
||||
mod = imp.new_module(fullname)
|
||||
sys.modules[fullname] = mod
|
||||
mod.__file__ = "<%s>" % self.__class__.__name__
|
||||
mod.__loader__ = self
|
||||
if ispkg:
|
||||
mod.__path__ = self._get__path__()
|
||||
exec code in mod.__dict__
|
||||
return mod
|
||||
|
||||
|
||||
class MetaImporter(TestImporter):
|
||||
def _get__path__(self):
|
||||
return []
|
||||
|
||||
class PathImporter(TestImporter):
|
||||
def _get__path__(self):
|
||||
return [self.path]
|
||||
|
||||
|
||||
class ImportBlocker:
|
||||
"""Place an ImportBlocker instance on sys.meta_path and you
|
||||
can be sure the modules you specified can't be imported, even
|
||||
if it's a builtin."""
|
||||
def __init__(self, *namestoblock):
|
||||
self.namestoblock = dict.fromkeys(namestoblock)
|
||||
def find_module(self, fullname, path=None):
|
||||
if fullname in self.namestoblock:
|
||||
return self
|
||||
return None
|
||||
def load_module(self, fullname):
|
||||
raise ImportError, "I dare you"
|
||||
|
||||
|
||||
class ImpWrapper:
|
||||
|
||||
def __init__(self, path=None):
|
||||
if path is not None and not os.path.isdir(path):
|
||||
raise ImportError
|
||||
self.path = path
|
||||
|
||||
def find_module(self, fullname, path=None):
|
||||
subname = fullname.split(".")[-1]
|
||||
if subname != fullname and self.path is None:
|
||||
return None
|
||||
if self.path is None:
|
||||
path = None
|
||||
else:
|
||||
path = [self.path]
|
||||
try:
|
||||
file, filename, stuff = imp.find_module(subname, path)
|
||||
except ImportError:
|
||||
return None
|
||||
return ImpLoader(file, filename, stuff)
|
||||
|
||||
|
||||
class ImpLoader:
|
||||
|
||||
def __init__(self, file, filename, stuff):
|
||||
self.file = file
|
||||
self.filename = filename
|
||||
self.stuff = stuff
|
||||
|
||||
def load_module(self, fullname):
|
||||
mod = imp.load_module(fullname, self.file, self.filename, self.stuff)
|
||||
if self.file:
|
||||
self.file.close()
|
||||
mod.__loader__ = self # for introspection
|
||||
return mod
|
||||
|
||||
|
||||
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.tracker = ImportTracker()
|
||||
sys.meta_path.insert(0, self.tracker)
|
||||
|
||||
def tearDown(self):
|
||||
sys.path[:] = self.path
|
||||
sys.meta_path[:] = self.meta_path
|
||||
sys.path_hooks[:] = self.path_hooks
|
||||
sys.path_importer_cache.clear()
|
||||
for fullname in self.tracker.imports:
|
||||
if fullname in sys.modules:
|
||||
del sys.modules[fullname]
|
||||
|
||||
|
||||
class ImportHooksTestCase(ImportHooksBaseTestCase):
|
||||
|
||||
def doTestImports(self, importer=None):
|
||||
import hooktestmodule
|
||||
import hooktestpackage
|
||||
import hooktestpackage.sub
|
||||
import hooktestpackage.sub.subber
|
||||
self.assertEqual(hooktestmodule.get_name(),
|
||||
"hooktestmodule")
|
||||
self.assertEqual(hooktestpackage.get_name(),
|
||||
"hooktestpackage")
|
||||
self.assertEqual(hooktestpackage.sub.get_name(),
|
||||
"hooktestpackage.sub")
|
||||
self.assertEqual(hooktestpackage.sub.subber.get_name(),
|
||||
"hooktestpackage.sub.subber")
|
||||
if importer:
|
||||
self.assertEqual(hooktestmodule.__loader__, importer)
|
||||
self.assertEqual(hooktestpackage.__loader__, importer)
|
||||
self.assertEqual(hooktestpackage.sub.__loader__, importer)
|
||||
self.assertEqual(hooktestpackage.sub.subber.__loader__, importer)
|
||||
|
||||
def testMetaPath(self):
|
||||
i = MetaImporter()
|
||||
sys.meta_path.append(i)
|
||||
self.doTestImports(i)
|
||||
|
||||
def testPathHook(self):
|
||||
sys.path_hooks.append(PathImporter)
|
||||
sys.path.append(test_path)
|
||||
self.doTestImports()
|
||||
|
||||
def testBlocker(self):
|
||||
mname = "exceptions" # an arbitrary harmless builtin module
|
||||
if mname in sys.modules:
|
||||
del sys.modules[mname]
|
||||
sys.meta_path.append(ImportBlocker(mname))
|
||||
try:
|
||||
__import__(mname)
|
||||
except ImportError:
|
||||
pass
|
||||
else:
|
||||
self.fail("'%s' was not supposed to be importable" % mname)
|
||||
|
||||
def testImpWrapper(self):
|
||||
i = ImpWrapper()
|
||||
sys.meta_path.append(i)
|
||||
sys.path_hooks.append(ImpWrapper)
|
||||
mnames = ("colorsys", "urlparse", "distutils.core", "compiler.misc")
|
||||
for mname in mnames:
|
||||
parent = mname.split(".")[0]
|
||||
for n in sys.modules.keys():
|
||||
if n.startswith(parent):
|
||||
del sys.modules[n]
|
||||
for mname in mnames:
|
||||
m = __import__(mname, globals(), locals(), ["__dummy__"])
|
||||
m.__loader__ # to make sure we actually handled the import
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_support.run_unittest(ImportHooksTestCase)
|
180
Lib/test/test_zipimport.py
Normal file
180
Lib/test/test_zipimport.py
Normal file
|
@ -0,0 +1,180 @@
|
|||
import sys
|
||||
import os
|
||||
import marshal
|
||||
import imp
|
||||
import struct
|
||||
import time
|
||||
|
||||
import zlib # implied prerequisite
|
||||
from zipfile import ZipFile, ZipInfo, ZIP_STORED, ZIP_DEFLATED
|
||||
from test import test_support
|
||||
from test.test_importhooks import ImportHooksBaseTestCase, test_src, test_co
|
||||
|
||||
import zipimport
|
||||
|
||||
|
||||
def make_pyc(co, mtime):
|
||||
data = marshal.dumps(co)
|
||||
pyc = imp.get_magic() + struct.pack("<i", mtime) + data
|
||||
return pyc
|
||||
|
||||
NOW = time.time()
|
||||
test_pyc = make_pyc(test_co, NOW)
|
||||
|
||||
|
||||
if __debug__:
|
||||
pyc_ext = ".pyc"
|
||||
else:
|
||||
pyc_ext = ".pyo"
|
||||
|
||||
|
||||
TESTMOD = "ziptestmodule"
|
||||
TESTPACK = "ziptestpackage"
|
||||
TEMP_ZIP = "junk95142.zip"
|
||||
|
||||
|
||||
class UncompressedZipImportTestCase(ImportHooksBaseTestCase):
|
||||
|
||||
compression = ZIP_STORED
|
||||
|
||||
def setUp(self):
|
||||
# We're reusing the zip archive path, so we must clear the
|
||||
# cached directory info.
|
||||
zipimport._zip_directory_cache.clear()
|
||||
ImportHooksBaseTestCase.setUp(self)
|
||||
|
||||
def doTest(self, expected_ext, files, *modules):
|
||||
z = ZipFile(TEMP_ZIP, "w")
|
||||
try:
|
||||
for name, (mtime, data) in files.items():
|
||||
zinfo = ZipInfo(name, time.localtime(mtime))
|
||||
zinfo.compress_type = self.compression
|
||||
z.writestr(zinfo, data)
|
||||
z.close()
|
||||
sys.path.insert(0, TEMP_ZIP)
|
||||
|
||||
mod = __import__(".".join(modules), globals(), locals(),
|
||||
["__dummy__"])
|
||||
file = mod.get_file()
|
||||
self.assertEquals(file, os.path.join(TEMP_ZIP,
|
||||
os.sep.join(modules) + expected_ext))
|
||||
finally:
|
||||
z.close()
|
||||
os.remove(TEMP_ZIP)
|
||||
|
||||
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.makeSuite() does.)
|
||||
#
|
||||
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 testBadMagic(self):
|
||||
# make pyc magic word invalid, forcing loading from .py
|
||||
m0 = ord(test_pyc[0])
|
||||
m0 ^= 0x04 # flip an arbitrary bit
|
||||
badmagic_pyc = chr(m0) + test_pyc[1:]
|
||||
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
|
||||
m0 = ord(test_pyc[0])
|
||||
m0 ^= 0x04 # flip an arbitrary bit
|
||||
badmagic_pyc = chr(m0) + test_pyc[1:]
|
||||
files = {TESTMOD + pyc_ext: (NOW, badmagic_pyc)}
|
||||
try:
|
||||
self.doTest(".py", files, TESTMOD)
|
||||
except ImportError:
|
||||
pass
|
||||
else:
|
||||
self.fail("expected ImportError; import from bad pyc")
|
||||
|
||||
def testBadMTime(self):
|
||||
t3 = ord(test_pyc[7])
|
||||
t3 ^= 0x02 # flip the second bit -- not the first as that one
|
||||
# isn't stored in the .py's mtime in the zip archive.
|
||||
badtime_pyc = test_pyc[:7] + chr(t3) + test_pyc[8:]
|
||||
files = {TESTMOD + ".py": (NOW, test_src),
|
||||
TESTMOD + pyc_ext: (NOW, badtime_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 testDeepPackage(self):
|
||||
packdir = TESTPACK + os.sep
|
||||
packdir2 = packdir + packdir
|
||||
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, TESTPACK, TESTMOD)
|
||||
|
||||
def testGetData(self):
|
||||
z = ZipFile(TEMP_ZIP, "w")
|
||||
z.compression = self.compression
|
||||
try:
|
||||
name = "testdata.dat"
|
||||
data = "".join([chr(x) for x in range(256)]) * 500
|
||||
z.writestr(name, data)
|
||||
z.close()
|
||||
zi = zipimport.zipimporter(TEMP_ZIP)
|
||||
self.assertEquals(data, zi.get_data(name))
|
||||
finally:
|
||||
z.close()
|
||||
os.remove(TEMP_ZIP)
|
||||
|
||||
def testImporterAttr(self):
|
||||
src = """if 1: # indent hack
|
||||
def get_file():
|
||||
return __file__
|
||||
if __importer__.get_data("some.data") != "some data":
|
||||
raise AssertionError, "bad data"\n"""
|
||||
pyc = make_pyc(compile(src, "<???>", "exec"), NOW)
|
||||
files = {TESTMOD + pyc_ext: (NOW, pyc),
|
||||
"some.data": (NOW, "some data")}
|
||||
self.doTest(pyc_ext, files, TESTMOD)
|
||||
|
||||
|
||||
class CompressedZipImportTestCase(UncompressedZipImportTestCase):
|
||||
compression = ZIP_DEFLATED
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_support.run_unittest(UncompressedZipImportTestCase)
|
||||
test_support.run_unittest(CompressedZipImportTestCase)
|
Loading…
Add table
Add a link
Reference in a new issue