Introduce importlib.abc. The module contains various ABCs related to imports

(mostly stuff specified by PEP 302). There are two ABCs, PyLoader and
PyPycLoader, which help with implementing source and source/bytecode loaders by
implementing load_module in terms of other methods. This removes a lot of
gritty details loaders typically have to worry about.

3 files changed, 404 insertions, 96 deletions

diff --git a/Lib/importlib/test/source/test_abc_loader.py b/Lib/importlib/test/source/test_abc_loader.py
new file mode 100644
index 0000000..c937793
--- /dev/null
+++ b/Lib/importlib/test/source/test_abc_loader.py
@@ -0,0 +1,390 @@
+import importlib
+from importlib import abc
+from .. import abc as testing_abc
+from .. import util
+from . import util as source_util
+import imp
+import marshal
+import os
+import sys
+import types
+import unittest
+
+
+class PyLoaderMock(abc.PyLoader):
+
+    # Globals that should be defined for all modules.
+    source = ("_ = '::'.join([__name__, __file__, __package__, "
+              "repr(__loader__)])")
+
+    def __init__(self, data):
+        """Take a dict of 'module_name: path' pairings.
+
+        Paths should have no file extension, allowing packages to be denoted by
+        ending in '__init__'.
+
+        """
+        self.module_paths = data
+        self.path_to_module = {val:key for key,val in data.items()}
+
+    def get_data(self, path):
+        if path not in self.path_to_module:
+            raise IOError
+        return self.source.encode('utf-8')
+
+    def is_package(self, name):
+        try:
+            return '__init__' in self.module_paths[name]
+        except KeyError:
+            raise ImportError
+
+    def get_source(self, name):  # Should not be needed.
+        raise NotImplementedError
+
+    def source_path(self, name):
+        try:
+            return self.module_paths[name]
+        except KeyError:
+            raise ImportError
+
+
+class PyPycLoaderMock(abc.PyPycLoader, PyLoaderMock):
+
+    default_mtime = 1
+
+    def __init__(self, source, bc={}):
+        """Initialize mock.
+
+        'bc' is a dict keyed on a module's name. The value is dict with
+        possible keys of 'path', 'mtime', 'magic', and 'bc'. Except for 'path',
+        each of those keys control if any part of created bytecode is to
+        deviate from default values.
+
+        """
+        super().__init__(source)
+        self.module_bytecode = {}
+        self.path_to_bytecode = {}
+        self.bytecode_to_path = {}
+        for name, data in bc.items():
+            self.path_to_bytecode[data['path']] = name
+            self.bytecode_to_path[name] = data['path']
+            magic = data.get('magic', imp.get_magic())
+            mtime = importlib._w_long(data.get('mtime', self.default_mtime))
+            if 'bc' in data:
+                bc = data['bc']
+            else:
+                bc = self.compile_bc(name)
+            self.module_bytecode[name] = magic + mtime + bc
+
+    def compile_bc(self, name):
+        source_path = self.module_paths.get(name, '<test>') or '<test>'
+        code = compile(self.source, source_path, 'exec')
+        return marshal.dumps(code)
+
+    def source_mtime(self, name):
+        if name in self.module_paths:
+            return self.default_mtime
+        elif name in self.module_bytecode:
+            return None
+        else:
+            raise ImportError
+
+    def bytecode_path(self, name):
+        try:
+            return self.bytecode_to_path[name]
+        except KeyError:
+            if name in self.module_paths:
+                return None
+            else:
+                raise ImportError
+
+    def write_bytecode(self, name, bytecode):
+        self.module_bytecode[name] = bytecode
+        return True
+
+    def get_data(self, path):
+        if path in self.path_to_module:
+            return super().get_data(path)
+        elif path in self.path_to_bytecode:
+            name = self.path_to_bytecode[path]
+            return self.module_bytecode[name]
+        else:
+            raise IOError
+
+    def is_package(self, name):
+        try:
+            return super().is_package(name)
+        except TypeError:
+            return '__init__' in self.bytecode_to_path[name]
+
+
+class PyLoaderTests(testing_abc.LoaderTests):
+
+    """Tests for importlib.abc.PyLoader."""
+
+    mocker = PyLoaderMock
+
+    def eq_attrs(self, ob, **kwargs):
+        for attr, val in kwargs.items():
+            self.assertEqual(getattr(ob, attr), val)
+
+    def test_module(self):
+        name = '<module>'
+        path = 'path/to/module'
+        mock = self.mocker({name: path})
+        with util.uncache(name):
+            module = mock.load_module(name)
+            self.assert_(name in sys.modules)
+        self.eq_attrs(module, __name__=name, __file__=path, __package__='',
+                        __loader__=mock)
+        self.assert_(not hasattr(module, '__path__'))
+        return mock, name
+
+    def test_package(self):
+        name = '<pkg>'
+        path = '/path/to/<pkg>/__init__'
+        mock = self.mocker({name: path})
+        with util.uncache(name):
+            module = mock.load_module(name)
+            self.assert_(name in sys.modules)
+        self.eq_attrs(module, __name__=name, __file__=path,
+                __path__=[os.path.dirname(path)], __package__=name,
+                __loader__=mock)
+        return mock, name
+
+    def test_lacking_parent(self):
+        name = 'pkg.mod'
+        path = 'path/to/pkg/mod'
+        mock = self.mocker({name: path})
+        with util.uncache(name):
+            module = mock.load_module(name)
+            self.assert_(name in sys.modules)
+        self.eq_attrs(module, __name__=name, __file__=path, __package__='pkg',
+                        __loader__=mock)
+        self.assert_(not hasattr(module, '__path__'))
+        return mock, name
+
+    def test_module_reuse(self):
+        name = 'mod'
+        path = 'path/to/mod'
+        module = imp.new_module(name)
+        mock = self.mocker({name: path})
+        with util.uncache(name):
+            sys.modules[name] = module
+            loaded_module = mock.load_module(name)
+            self.assert_(loaded_module is module)
+            self.assert_(sys.modules[name] is module)
+        return mock, name
+
+    def test_state_after_failure(self):
+        name = "mod"
+        module = imp.new_module(name)
+        module.blah = None
+        mock = self.mocker({name: 'path/to/mod'})
+        mock.source = "1/0"
+        with util.uncache(name):
+            sys.modules[name] = module
+            self.assertRaises(ZeroDivisionError, mock.load_module, name)
+            self.assert_(sys.modules[name] is module)
+            self.assert_(hasattr(module, 'blah'))
+        return mock
+
+    def test_unloadable(self):
+        name = "mod"
+        mock = self.mocker({name: 'path/to/mod'})
+        mock.source = "1/0"
+        with util.uncache(name):
+            self.assertRaises(ZeroDivisionError, mock.load_module, name)
+            self.assert_(name not in sys.modules)
+        return mock
+
+
+class PyLoaderInterfaceTests(unittest.TestCase):
+
+
+    def test_no_source_path(self):
+        # No source path should lead to ImportError.
+        name = 'mod'
+        mock = PyLoaderMock({})
+        with util.uncache(name):
+            self.assertRaises(ImportError, mock.load_module, name)
+
+    def test_source_path_is_None(self):
+        name = 'mod'
+        mock = PyLoaderMock({name: None})
+        with util.uncache(name):
+            self.assertRaises(ImportError, mock.load_module, name)
+
+
+class PyPycLoaderTests(PyLoaderTests):
+
+    """Tests for importlib.abc.PyPycLoader."""
+
+    mocker = PyPycLoaderMock
+
+    @source_util.writes_bytecode
+    def verify_bytecode(self, mock, name):
+        assert name in mock.module_paths
+        self.assert_(name in mock.module_bytecode)
+        magic = mock.module_bytecode[name][:4]
+        self.assertEqual(magic, imp.get_magic())
+        mtime = importlib._r_long(mock.module_bytecode[name][4:8])
+        self.assertEqual(mtime, 1)
+        bc = mock.module_bytecode[name][8:]
+
+
+    def test_module(self):
+        mock, name = super().test_module()
+        self.verify_bytecode(mock, name)
+
+    def test_package(self):
+        mock, name = super().test_package()
+        self.verify_bytecode(mock, name)
+
+    def test_lacking_parent(self):
+        mock, name = super().test_lacking_parent()
+        self.verify_bytecode(mock, name)
+
+    def test_module_reuse(self):
+        mock, name = super().test_module_reuse()
+        self.verify_bytecode(mock, name)
+
+    def test_state_after_failure(self):
+        super().test_state_after_failure()
+
+    def test_unloadable(self):
+        super().test_unloadable()
+
+
+class SkipWritingBytecodeTests(unittest.TestCase):
+
+    """Test that bytecode is properly handled based on
+    sys.dont_write_bytecode."""
+
+    @source_util.writes_bytecode
+    def run_test(self, dont_write_bytecode):
+        name = 'mod'
+        mock = PyPycLoaderMock({name: 'path/to/mod'})
+        sys.dont_write_bytecode = dont_write_bytecode
+        with util.uncache(name):
+            mock.load_module(name)
+        self.assert_((name in mock.module_bytecode) is not
+                        dont_write_bytecode)
+
+    def test_no_bytecode_written(self):
+        self.run_test(True)
+
+    def test_bytecode_written(self):
+        self.run_test(False)
+
+
+class RegeneratedBytecodeTests(unittest.TestCase):
+
+    """Test that bytecode is regenerated as expected."""
+
+    @source_util.writes_bytecode
+    def test_different_magic(self):
+        # A different magic number should lead to new bytecode.
+        name = 'mod'
+        bad_magic = b'\x00\x00\x00\x00'
+        assert bad_magic != imp.get_magic()
+        mock = PyPycLoaderMock({name: 'path/to/mod'},
+                                {name: {'path': 'path/to/mod.bytecode',
+                                        'magic': bad_magic}})
+        with util.uncache(name):
+            mock.load_module(name)
+        self.assert_(name in mock.module_bytecode)
+        magic = mock.module_bytecode[name][:4]
+        self.assertEqual(magic, imp.get_magic())
+
+    @source_util.writes_bytecode
+    def test_old_mtime(self):
+        # Bytecode with an older mtime should be regenerated.
+        name = 'mod'
+        old_mtime = PyPycLoaderMock.default_mtime - 1
+        mock = PyPycLoaderMock({name: 'path/to/mod'},
+                {name: {'path': 'path/to/mod.bytecode', 'mtime': old_mtime}})
+        with util.uncache(name):
+            mock.load_module(name)
+        self.assert_(name in mock.module_bytecode)
+        mtime = importlib._r_long(mock.module_bytecode[name][4:8])
+        self.assertEqual(mtime, PyPycLoaderMock.default_mtime)
+
+
+class BadBytecodeFailureTests(unittest.TestCase):
+
+    """Test import failures when there is no source and parts of the bytecode
+    is bad."""
+
+    def test_bad_magic(self):
+        # A bad magic number should lead to an ImportError.
+        name = 'mod'
+        bad_magic = b'\x00\x00\x00\x00'
+        mock = PyPycLoaderMock({}, {name: {'path': 'path/to/mod',
+                                            'magic': bad_magic}})
+        with util.uncache(name):
+            self.assertRaises(ImportError, mock.load_module, name)
+
+    def test_bad_bytecode(self):
+        # Bad code object bytecode should elad to an ImportError.
+        name = 'mod'
+        mock = PyPycLoaderMock({}, {name: {'path': '/path/to/mod', 'bc': b''}})
+        with util.uncache(name):
+            self.assertRaises(ImportError, mock.load_module, name)
+
+
+def raise_ImportError(*args, **kwargs):
+    raise ImportError
+
+class MissingPathsTests(unittest.TestCase):
+
+    """Test what happens when a source or bytecode path does not exist (either
+    from *_path returning None or raising ImportError)."""
+
+    def test_source_path_None(self):
+        # Bytecode should be used when source_path returns None, along with
+        # __file__ being set to the bytecode path.
+        name = 'mod'
+        bytecode_path = 'path/to/mod'
+        mock = PyPycLoaderMock({name: None}, {name: {'path': bytecode_path}})
+        with util.uncache(name):
+            module = mock.load_module(name)
+        self.assertEqual(module.__file__, bytecode_path)
+
+    # Testing for bytecode_path returning None handled by all tests where no
+    # bytecode initially exists.
+
+    def test_all_paths_None(self):
+        # If all *_path methods return None, raise ImportError.
+        name = 'mod'
+        mock = PyPycLoaderMock({name: None})
+        with util.uncache(name):
+            self.assertRaises(ImportError, mock.load_module, name)
+
+    def test_source_path_ImportError(self):
+        # An ImportError from source_path should trigger an ImportError.
+        name = 'mod'
+        mock = PyPycLoaderMock({}, {name: {'path': 'path/to/mod'}})
+        with util.uncache(name):
+            self.assertRaises(ImportError, mock.load_module, name)
+
+    def test_bytecode_path_ImportError(self):
+        # An ImportError from bytecode_path should trigger an ImportError.
+        name = 'mod'
+        mock = PyPycLoaderMock({name: 'path/to/mod'})
+        bad_meth = types.MethodType(raise_ImportError, mock)
+        mock.bytecode_path = bad_meth
+        with util.uncache(name):
+            self.assertRaises(ImportError, mock.load_module, name)
+
+
+def test_main():
+    from test.support import run_unittest
+    run_unittest(PyLoaderTests, PyLoaderInterfaceTests,
+                    PyPycLoaderTests, SkipWritingBytecodeTests,
+                    RegeneratedBytecodeTests, BadBytecodeFailureTests,
+                    MissingPathsTests)
+
+
+if __name__ == '__main__':
+    test_main()
diff --git a/Lib/importlib/test/source/test_loader.py b/Lib/importlib/test/source/test_file_loader.py
index 960210f..3da4426 100644
--- a/Lib/importlib/test/source/test_loader.py
+++ b/Lib/importlib/test/source/test_file_loader.py
@@ -102,102 +102,6 @@ class SimpleTest(unittest.TestCase):
             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
-
-    @source_util.writes_bytecode
-    def run_test(self, assertion):
-        with source_util.create_modules('_temp') as mapping:
-            loader = importlib.PyPycFileLoader('_temp', mapping['_temp'], False)
-            loader.load_module('_temp')
-            bytecode_path = source_util.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 source_util.create_modules('_temp') as mapping:
-            py_compile.compile(mapping['_temp'])
-            os.unlink(mapping['_temp'])
-            bytecode_path = source_util.bytecode_path(mapping['_temp'])
-            with open(bytecode_path, 'r+b') as file:
-                file.seek(0)
-                file.write(b'\x00\x00\x00\x00')
-            loader = importlib.PyPycFileLoader('_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.PyPycFileLoader(module, file, pkg)
-        return loader.load_module(module)
-
-    def run_test(self, test, *create, pkg=False):
-        create += (test,)
-        with source_util.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.PyPycFileLoader(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
diff --git a/Lib/importlib/test/source/util.py b/Lib/importlib/test/source/util.py
index f02d491..280edb4 100644
--- a/Lib/importlib/test/source/util.py
+++ b/Lib/importlib/test/source/util.py
@@ -1,5 +1,6 @@
 from .. import util
 import contextlib
+import functools
 import imp
 import os
 import os.path
@@ -9,11 +10,24 @@ from test import support
 
 
 def writes_bytecode(fxn):
+    """Decorator to protect sys.dont_write_bytecode from mutation."""
+    @functools.wraps(fxn)
+    def wrapper(*args, **kwargs):
+        original = sys.dont_write_bytecode
+        sys.dont_write_bytecode = False
+        to_return = fxn(*args, **kwargs)
+        sys.dont_write_bytecode = original
+        return to_return
+    return wrapper
+
+
+def writes_bytecode_files(fxn):
     """Decorator that returns the function if writing bytecode is enabled, else
     a stub function that accepts anything and simply returns None."""
     if sys.dont_write_bytecode:
         return lambda *args, **kwargs: None
     else:
+        @functools.wraps(fxn)
         def wrapper(*args, **kwargs):
             to_return = fxn(*args, **kwargs)
             sys.dont_write_bytecode = False