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.
diff --git a/Lib/test/test_importhooks.py b/Lib/test/test_importhooks.py
new file mode 100644
index 0000000..070b578
--- /dev/null
+++ b/Lib/test/test_importhooks.py
@@ -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)
diff --git a/Lib/test/test_zipimport.py b/Lib/test/test_zipimport.py
new file mode 100644
index 0000000..07520a7
--- /dev/null
+++ b/Lib/test/test_zipimport.py
@@ -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)