Add initial implementation of importlib. See the NOTES files for what is

planned for the package.

There are no docs yet, but they are coming once the API for the first new
function, importlib.import_module() is finalized.

7 files changed, 617 insertions, 0 deletions

diff --git a/Lib/importlib/test/source/__init__.py b/Lib/importlib/test/source/__init__.py
new file mode 100644
index 0000000..8d7c49d
--- /dev/null
+++ b/Lib/importlib/test/source/__init__.py
@@ -0,0 +1,13 @@
+import importlib.test
+import os.path
+import unittest
+
+
+def test_suite():
+    directory = os.path.dirname(__file__)
+    return importlib.test.test_suite('importlib.test.source', directory)
+
+
+if __name__ == '__main__':
+    from test.support import run_unittest
+    run_unittest(test_suite())
diff --git a/Lib/importlib/test/source/test_case_sensitivity.py b/Lib/importlib/test/source/test_case_sensitivity.py
new file mode 100644
index 0000000..1a5ff2f
--- /dev/null
+++ b/Lib/importlib/test/source/test_case_sensitivity.py
@@ -0,0 +1,57 @@
+"""Test case-sensitivity (PEP 235)."""
+import importlib
+from .. import support
+import os
+import sys
+from test import support as test_support
+import unittest
+
+
+class CaseSensitivityTest(unittest.TestCase):
+
+    """PEP 235 dictates that on case-preserving, case-insensitive file systems
+    that imports are case-sensitive unless the PYTHONCASEOK environment
+    variable is set."""
+
+    name = 'MoDuLe'
+    assert name != name.lower()
+
+    def find(self, path):
+        finder = importlib.PyFileImporter(path)
+        return finder.find_module(self.name)
+
+    def sensitivity_test(self):
+        """Look for a module with matching and non-matching sensitivity."""
+        sensitive_pkg = 'sensitive.{0}'.format(self.name)
+        insensitive_pkg = 'insensitive.{0}'.format(self.name.lower())
+        with support.create_modules(insensitive_pkg, sensitive_pkg) as mapping:
+            sensitive_path = os.path.join(mapping['.root'], 'sensitive')
+            insensitive_path = os.path.join(mapping['.root'], 'insensitive')
+            return self.find(sensitive_path), self.find(insensitive_path)
+
+    def test_sensitive(self):
+        with test_support.EnvironmentVarGuard() as env:
+            env.unset('PYTHONCASEOK')
+            sensitive, insensitive = self.sensitivity_test()
+            self.assert_(hasattr(sensitive, 'load_module'))
+            self.assert_(self.name in sensitive._base_path)
+            self.assert_(insensitive is None)
+
+    def test_insensitive(self):
+        with test_support.EnvironmentVarGuard() as env:
+            env.set('PYTHONCASEOK', '1')
+            sensitive, insensitive = self.sensitivity_test()
+            self.assert_(hasattr(sensitive, 'load_module'))
+            self.assert_(self.name in sensitive._base_path)
+            self.assert_(hasattr(insensitive, 'load_module'))
+            self.assert_(self.name in insensitive._base_path)
+
+
+def test_main():
+    if sys.platform not in ('win32', 'darwin', 'cygwin'):
+        return
+    test_support.run_unittest(CaseSensitivityTest)
+
+
+if __name__ == '__main__':
+    test_main()
diff --git a/Lib/importlib/test/source/test_finder.py b/Lib/importlib/test/source/test_finder.py
new file mode 100644
index 0000000..cf80799
--- /dev/null
+++ b/Lib/importlib/test/source/test_finder.py
@@ -0,0 +1,130 @@
+import importlib
+from .. import finder_tests
+from .. import support
+import os
+import py_compile
+import unittest
+import warnings
+
+
+class FinderTests(finder_tests.FinderTests):
+
+    """For a top-level module, it should just be found directly in the
+    directory being searched. This is true for a directory with source
+    [top-level source], bytecode [top-level bc], or both [top-level both].
+    There is also the possibility that it is a package [top-level package], in
+    which case there will be a directory with the module name and an
+    __init__.py file. If there is a directory without an __init__.py an
+    ImportWarning is returned [empty dir].
+
+    For sub-modules and sub-packages, the same happens as above but only use
+    the tail end of the name [sub module] [sub package] [sub empty].
+
+    When there is a conflict between a package and module having the same name
+    in the same directory, the package wins out [package over module]. This is
+    so that imports of modules within the package can occur rather than trigger
+    an import error.
+
+    When there is a package and module with the same name, always pick the
+    package over the module [package over module]. This is so that imports from
+    the package have the possibility of succeeding.
+
+    """
+
+    def import_(self, root, module):
+        finder = importlib.PyFileImporter(root)
+        return finder.find_module(module)
+
+    def run_test(self, test, create=None, *, compile_=None, unlink=None):
+        """Test the finding of 'test' with the creation of modules listed in
+        'create'.
+
+        Any names listed in 'compile_' are byte-compiled. Modules
+        listed in 'unlink' have their source files deleted.
+
+        """
+        if create is None:
+            create = {test}
+        with support.create_modules(*create) as mapping:
+            if compile_:
+                for name in compile_:
+                    py_compile.compile(mapping[name])
+            if unlink:
+                for name in unlink:
+                    os.unlink(mapping[name])
+            loader = self.import_(mapping['.root'], test)
+            self.assert_(hasattr(loader, 'load_module'))
+            return loader
+
+    def test_module(self):
+        # [top-level source]
+        self.run_test('top_level')
+        # [top-level bc]
+        self.run_test('top_level', compile_={'top_level'}, unlink={'top_level'})
+        # [top-level both]
+        self.run_test('top_level', compile_={'top_level'})
+
+    # [top-level package]
+    def test_package(self):
+        # Source.
+        self.run_test('pkg', {'pkg.__init__'})
+        # Bytecode.
+        self.run_test('pkg', {'pkg.__init__'}, compile_={'pkg.__init__'},
+                unlink={'pkg.__init__'})
+        # Both.
+        self.run_test('pkg', {'pkg.__init__'}, compile_={'pkg.__init__'})
+
+    # [sub module]
+    def test_module_in_package(self):
+        with support.create_modules('pkg.__init__', 'pkg.sub') as mapping:
+            pkg_dir = os.path.dirname(mapping['pkg.__init__'])
+            loader = self.import_(pkg_dir, 'pkg.sub')
+            self.assert_(hasattr(loader, 'load_module'))
+
+    # [sub package]
+    def test_package_in_package(self):
+        context = support.create_modules('pkg.__init__', 'pkg.sub.__init__')
+        with context as mapping:
+            pkg_dir = os.path.dirname(mapping['pkg.__init__'])
+            loader = self.import_(pkg_dir, 'pkg.sub')
+            self.assert_(hasattr(loader, 'load_module'))
+
+    # [sub empty]
+    def test_empty_sub_directory(self):
+        context = support.create_modules('pkg.__init__', 'pkg.sub.__init__')
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", ImportWarning)
+            with context as mapping:
+                os.unlink(mapping['pkg.sub.__init__'])
+                pkg_dir = os.path.dirname(mapping['pkg.__init__'])
+                self.assertRaises(ImportWarning, self.import_, pkg_dir,
+                                    'pkg.sub')
+
+    # [package over modules]
+    def test_package_over_module(self):
+        # XXX This is not a blackbox test!
+        name = '_temp'
+        loader = self.run_test(name, {'{0}.__init__'.format(name), name})
+        self.assert_('__init__' in loader._base_path)
+
+
+    def test_failure(self):
+        with support.create_modules('blah') as mapping:
+            nothing = self.import_(mapping['.root'], 'sdfsadsadf')
+            self.assert_(nothing is None)
+
+    # [empty dir]
+    def test_empty_dir(self):
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", ImportWarning)
+            self.assertRaises(ImportWarning, self.run_test, 'pkg',
+            {'pkg.__init__'}, unlink={'pkg.__init__'})
+
+
+def test_main():
+    from test.support import run_unittest
+    run_unittest(FinderTests)
+
+
+if __name__ == '__main__':
+    test_main()
diff --git a/Lib/importlib/test/source/test_loader.py b/Lib/importlib/test/source/test_loader.py
new file mode 100644
index 0000000..249bdc0
--- /dev/null
+++ b/Lib/importlib/test/source/test_loader.py
@@ -0,0 +1,201 @@
+import importlib
+from .. import support
+
+import imp
+import os
+import py_compile
+import sys
+import unittest
+
+
+class SimpleTest(unittest.TestCase):
+
+    """Should have no issue importing a source module [basic]. And if there is
+    a syntax error, it should raise a SyntaxError [syntax error].
+
+    """
+
+    # [basic]
+    def test_basic(self):
+        with support.create_modules('_temp') as mapping:
+            loader = importlib._PyFileLoader('_temp', mapping['_temp'], False)
+            loader.load_module('_temp')
+            self.assert_('_temp' in sys.modules)
+
+    # [syntax error]
+    def test_bad_syntax(self):
+        with support.create_modules('_temp') as mapping:
+            with open(mapping['_temp'], 'w') as file:
+                file.write('=')
+            loader = importlib._PyFileLoader('_temp', mapping['_temp'], False)
+            self.assertRaises(SyntaxError, loader.load_module, '_temp')
+            self.assert_('_temp' not in sys.modules)
+
+
+class DontWriteBytecodeTest(unittest.TestCase):
+
+    """If sys.dont_write_bytcode is true then no bytecode should be created."""
+
+    def tearDown(self):
+        sys.dont_write_bytecode = False
+
+    @support.writes_bytecode
+    def run_test(self, assertion):
+        with support.create_modules('_temp') as mapping:
+            loader = importlib._PyFileLoader('_temp', mapping['_temp'], False)
+            loader.load_module('_temp')
+            bytecode_path = support.bytecode_path(mapping['_temp'])
+            assertion(bytecode_path)
+
+    def test_bytecode_written(self):
+        fxn = lambda bc_path: self.assert_(os.path.exists(bc_path))
+        self.run_test(fxn)
+
+    def test_bytecode_not_written(self):
+        sys.dont_write_bytecode = True
+        fxn = lambda bc_path: self.assert_(not os.path.exists(bc_path))
+        self.run_test(fxn)
+
+
+class BadDataTest(unittest.TestCase):
+
+    """If the bytecode has a magic number that does not match the
+    interpreters', ImportError is raised [bad magic]. The timestamp can have
+    any value. And bad marshal data raises ValueError.
+
+    """
+
+    # [bad magic]
+    def test_bad_magic(self):
+        with support.create_modules('_temp') as mapping:
+            py_compile.compile(mapping['_temp'])
+            os.unlink(mapping['_temp'])
+            bytecode_path = support.bytecode_path(mapping['_temp'])
+            with open(bytecode_path, 'r+b') as file:
+                file.seek(0)
+                file.write(b'\x00\x00\x00\x00')
+            loader = importlib._PyFileLoader('_temp', mapping['_temp'], False)
+            self.assertRaises(ImportError, loader.load_module, '_temp')
+            self.assert_('_temp' not in sys.modules)
+
+
+class SourceBytecodeInteraction(unittest.TestCase):
+
+    """When both source and bytecode are present, certain rules dictate which
+    version of the code takes precedent. All things being equal, the bytecode
+    is used with the value of __file__ set to the source [basic top-level],
+    [basic package], [basic sub-module], [basic sub-package].
+
+    """
+
+    def import_(self, file, module, *, pkg=False):
+        loader = importlib._PyFileLoader(module, file, pkg)
+        return loader.load_module(module)
+
+    def run_test(self, test, *create, pkg=False):
+        create += (test,)
+        with support.create_modules(*create) as mapping:
+            for name in create:
+                py_compile.compile(mapping[name])
+            if pkg:
+                import_name = test.rsplit('.', 1)[0]
+            else:
+                import_name = test
+            loader = importlib._PyFileLoader(import_name, mapping[test], pkg)
+            # Because some platforms only have a granularity to the second for
+            # atime you can't check the physical files. Instead just make it an
+            # exception trigger if source was read.
+            loader.get_source = lambda self, x: 42
+            module = loader.load_module(import_name)
+            self.assertEqual(module.__file__, mapping[name])
+            self.assert_(import_name in sys.modules)
+            self.assertEqual(id(module), id(sys.modules[import_name]))
+
+    # [basic top-level]
+    def test_basic_top_level(self):
+        self.run_test('top_level')
+
+    # [basic package]
+    def test_basic_package(self):
+        self.run_test('pkg.__init__', pkg=True)
+
+    # [basic sub-module]
+    def test_basic_sub_module(self):
+        self.run_test('pkg.sub', 'pkg.__init__')
+
+    # [basic sub-package]
+    def test_basic_sub_package(self):
+        self.run_test('pkg.sub.__init__', 'pkg.__init__', pkg=True)
+
+
+class BadBytecodeTest(unittest.TestCase):
+
+    """But there are several things about the bytecode which might lead to the
+    source being preferred. If the magic number differs from what the
+    interpreter uses, then the source is used with the bytecode regenerated.
+    If the timestamp is older than the modification time for the source then
+    the bytecode is not used [bad timestamp].
+
+    But if the marshal data is bad, even if the magic number and timestamp
+    work, a ValueError is raised and the source is not used [bad marshal].
+
+    """
+
+    def import_(self, file, module_name):
+        loader = importlib._PyFileLoader(module_name, file, False)
+        module = loader.load_module(module_name)
+        self.assert_(module_name in sys.modules)
+
+    # [bad magic]
+    @support.writes_bytecode
+    def test_bad_magic(self):
+        with support.create_modules('_temp') as mapping:
+            py_compile.compile(mapping['_temp'])
+            bytecode_path = support.bytecode_path(mapping['_temp'])
+            with open(bytecode_path, 'r+b') as bytecode_file:
+                bytecode_file.seek(0)
+                bytecode_file.write(b'\x00\x00\x00\x00')
+            self.import_(mapping['_temp'], '_temp')
+            with open(bytecode_path, 'rb') as bytecode_file:
+                self.assertEqual(bytecode_file.read(4), imp.get_magic())
+
+    # [bad timestamp]
+    @support.writes_bytecode
+    def test_bad_bytecode(self):
+        zeros = b'\x00\x00\x00\x00'
+        with support.create_modules('_temp') as mapping:
+            py_compile.compile(mapping['_temp'])
+            bytecode_path = support.bytecode_path(mapping['_temp'])
+            with open(bytecode_path, 'r+b') as bytecode_file:
+                bytecode_file.seek(4)
+                bytecode_file.write(zeros)
+            self.import_(mapping['_temp'], '_temp')
+            source_mtime = os.path.getmtime(mapping['_temp'])
+            source_timestamp = importlib._w_long(source_mtime)
+            with open(bytecode_path, 'rb') as bytecode_file:
+                bytecode_file.seek(4)
+                self.assertEqual(bytecode_file.read(4), source_timestamp)
+
+    # [bad marshal]
+    def test_bad_marshal(self):
+        with support.create_modules('_temp') as mapping:
+            bytecode_path = support.bytecode_path(mapping['_temp'])
+            source_mtime = os.path.getmtime(mapping['_temp'])
+            source_timestamp = importlib._w_long(source_mtime)
+            with open(bytecode_path, 'wb') as bytecode_file:
+                bytecode_file.write(imp.get_magic())
+                bytecode_file.write(source_timestamp)
+                bytecode_file.write(b'AAAA')
+            self.assertRaises(ValueError, self.import_, mapping['_temp'],
+                                '_temp')
+            self.assert_('_temp' not in sys.modules)
+
+
+def test_main():
+    from test.support import run_unittest
+    run_unittest(SimpleTest, DontWriteBytecodeTest, BadDataTest,
+                 SourceBytecodeInteraction, BadBytecodeTest)
+
+
+if __name__ == '__main__':
+    test_main()
diff --git a/Lib/importlib/test/source/test_path_hook.py b/Lib/importlib/test/source/test_path_hook.py
new file mode 100644
index 0000000..ef410779
--- /dev/null
+++ b/Lib/importlib/test/source/test_path_hook.py
@@ -0,0 +1,23 @@
+import importlib
+from .. import support
+import unittest
+
+
+class PathHookTest(unittest.TestCase):
+
+    """Test the path hook for source."""
+
+    def test_success(self):
+        # XXX Only work on existing directories?
+        with support.create_modules('dummy') as mapping:
+            self.assert_(hasattr(importlib.FileImporter(mapping['.root']),
+                                 'find_module'))
+
+
+def test_main():
+    from test.support import run_unittest
+    run_unittest(PathHookTest)
+
+
+if __name__ == '__main__':
+    test_main()
diff --git a/Lib/importlib/test/source/test_reload.py b/Lib/importlib/test/source/test_reload.py
new file mode 100644
index 0000000..e123a28
--- /dev/null
+++ b/Lib/importlib/test/source/test_reload.py
@@ -0,0 +1,71 @@
+"""Test reload support.
+
+Reload support requires two things. One is that if module is loaded that
+already exists in sys.modules then it is reused. And two, if a reload fails the
+pre-existing module is left in a sane state.
+
+"""
+import imp
+import sys
+import types
+import unittest
+import importlib
+from .. import support
+
+
+class ReloadTests(unittest.TestCase):
+
+    name = '_temp'
+
+    def load_module(self, mapping):
+        return importlib._PyFileLoader(self.name, mapping[self.name], False)
+
+    def fake_mtime(self, fxn):
+        """Fake mtime to always be higher than expected."""
+        return lambda name: fxn(name) + 1
+
+    def test_module_reuse(self):
+        with support.create_modules(self.name) as mapping:
+            loader = self.load_module(mapping)
+            module = loader.load_module(self.name)
+            module_id = id(module)
+            module_dict_id = id(module.__dict__)
+            with open(mapping[self.name], 'w') as file:
+                file.write("testing_var = 42\n")
+            # For filesystems where the mtime is only to a second granularity,
+            # everything that has happened above can be too fast;
+            # force an mtime on the source that is guaranteed to be different
+            # than the original mtime.
+            loader.source_mtime = self.fake_mtime(loader.source_mtime)
+            module = loader.load_module(self.name)
+            self.assert_('testing_var' in module.__dict__,
+                         "'testing_var' not in "
+                            "{0}".format(list(module.__dict__.keys())))
+            self.assertEqual(module, sys.modules[self.name])
+            self.assertEqual(id(module), module_id)
+            self.assertEqual(id(module.__dict__), module_dict_id)
+
+    def test_bad_reload(self):
+        # A failed reload should leave the original module intact.
+        attributes = ('__file__', '__path__', '__package__')
+        value = '<test>'
+        with support.create_modules(self.name) as mapping:
+            orig_module = imp.new_module(self.name)
+            for attr in attributes:
+                setattr(orig_module, attr, value)
+            with open(mapping[self.name], 'w') as file:
+                file.write('+++ bad syntax +++')
+            loader = self.load_module(mapping)
+            self.assertRaises(SyntaxError, loader.load_module, self.name)
+            for attr in attributes:
+                self.assertEqual(getattr(orig_module, attr), value)
+
+
+
+def test_main():
+    from test.support import run_unittest
+    run_unittest(ReloadTests)
+
+
+if __name__ == '__main__':
+    test_main()
diff --git a/Lib/importlib/test/source/test_source_encoding.py b/Lib/importlib/test/source/test_source_encoding.py
new file mode 100644
index 0000000..97096df
--- /dev/null
+++ b/Lib/importlib/test/source/test_source_encoding.py
@@ -0,0 +1,122 @@
+import importlib
+from .. import support
+
+import codecs
+import re
+import sys
+# Because sys.path gets essentially blanked, need to have unicodedata already
+# imported for the parser to use.
+import unicodedata
+import unittest
+
+
+CODING_RE = re.compile(r'coding[:=]\s*([-\w.]+)')
+
+
+class EncodingTest(unittest.TestCase):
+
+    """PEP 3120 makes UTF-8 the default encoding for source code
+    [default encoding].
+
+    PEP 263 specifies how that can change on a per-file basis. Either the first
+    or second line can contain the encoding line [encoding first line]
+    encoding second line]. If the file has the BOM marker it is considered UTF-8
+    implicitly [BOM]. If any encoding is specified it must be UTF-8, else it is
+    an error [BOM and utf-8][BOM conflict].
+
+    """
+
+    variable = '\u00fc'
+    character = '\u00c9'
+    source_line = "{0} = '{1}'\n".format(variable, character)
+    module_name = '_temp'
+
+    def run_test(self, source):
+        with support.create_modules(self.module_name) as mapping:
+            with open(mapping[self.module_name], 'wb')as file:
+                file.write(source)
+            loader = importlib._PyFileLoader(self.module_name,
+                                       mapping[self.module_name], False)
+            return loader.load_module(self.module_name)
+
+    def create_source(self, encoding):
+        encoding_line = "# coding={0}".format(encoding)
+        assert CODING_RE.search(encoding_line)
+        source_lines = [encoding_line.encode('utf-8')]
+        source_lines.append(self.source_line.encode(encoding))
+        return b'\n'.join(source_lines)
+
+    def test_non_obvious_encoding(self):
+        # Make sure that an encoding that has never been a standard one for
+        # Python works.
+        encoding_line = "# coding=koi8-r"
+        assert CODING_RE.search(encoding_line)
+        source = "{0}\na=42\n".format(encoding_line).encode("koi8-r")
+        self.run_test(source)
+
+    # [default encoding]
+    def test_default_encoding(self):
+        self.run_test(self.source_line.encode('utf-8'))
+
+    # [encoding first line]
+    def test_encoding_on_first_line(self):
+        encoding = 'Latin-1'
+        source = self.create_source(encoding)
+        self.run_test(source)
+
+    # [encoding second line]
+    def test_encoding_on_second_line(self):
+        source = b"#/usr/bin/python\n" + self.create_source('Latin-1')
+        self.run_test(source)
+
+    # [BOM]
+    def test_bom(self):
+        self.run_test(codecs.BOM_UTF8 + self.source_line.encode('utf-8'))
+
+    # [BOM and utf-8]
+    def test_bom_and_utf_8(self):
+        source = codecs.BOM_UTF8 + self.create_source('utf-8')
+        self.run_test(source)
+
+    # [BOM conflict]
+    def test_bom_conflict(self):
+        source = codecs.BOM_UTF8 + self.create_source('latin-1')
+        self.assertRaises(SyntaxError, self.run_test, source)
+
+
+class LineEndingTest(unittest.TestCase):
+
+    r"""Source written with the three types of line endings (\n, \r\n, \r)
+    need to be readable [cr][crlf][lf]."""
+
+    def run_test(self, line_ending):
+        module_name = '_temp'
+        source_lines = [b"a = 42", b"b = -13", b'']
+        source = line_ending.join(source_lines)
+        with support.create_modules(module_name) as mapping:
+            with open(mapping[module_name], 'wb') as file:
+                file.write(source)
+            loader = importlib._PyFileLoader(module_name, mapping[module_name],
+                                                False)
+            return loader.load_module(module_name)
+
+    # [cr]
+    def test_cr(self):
+        self.run_test(b'\r')
+
+    # [crlf]
+    def test_crlf(self):
+        self.run_test(b'\r\n')
+
+    # [lf]
+    def test_lf(self):
+        self.run_test(b'\n')
+
+
+def test_main():
+    from test.support import run_unittest
+    run_unittest(EncodingTest, LineEndingTest)
+
+
+if __name__ == '__main__':
+    test_main()