Issue #19721: Consolidate test_importlib utility code into a single

module.

1 files changed, 142 insertions, 5 deletions

diff --git a/Lib/test/test_importlib/util.py b/Lib/test/test_importlib/util.py
index 885cec3..056293e 100644
--- a/Lib/test/test_importlib/util.py
+++ b/Lib/test/test_importlib/util.py
@@ -1,12 +1,49 @@
-from contextlib import contextmanager
-from importlib import util, invalidate_caches
+import builtins
+import contextlib
+import errno
+import functools
+import importlib
+from importlib import machinery, util, invalidate_caches
+import os
 import os.path
 from test import support
 import unittest
 import sys
+import tempfile
 import types
 
 
+BUILTINS = types.SimpleNamespace()
+BUILTINS.good_name = None
+BUILTINS.bad_name = None
+if 'errno' in sys.builtin_module_names:
+    BUILTINS.good_name = 'errno'
+if 'importlib' not in sys.builtin_module_names:
+    BUILTINS.bad_name = 'importlib'
+
+EXTENSIONS = types.SimpleNamespace()
+EXTENSIONS.path = None
+EXTENSIONS.ext = None
+EXTENSIONS.filename = None
+EXTENSIONS.file_path = None
+EXTENSIONS.name = '_testcapi'
+
+def _extension_details():
+    global EXTENSIONS
+    for path in sys.path:
+        for ext in machinery.EXTENSION_SUFFIXES:
+            filename = EXTENSIONS.name + ext
+            file_path = os.path.join(path, filename)
+            if os.path.exists(file_path):
+                EXTENSIONS.path = path
+                EXTENSIONS.ext = ext
+                EXTENSIONS.filename = filename
+                EXTENSIONS.file_path = file_path
+                return
+
+_extension_details()
+
+
 def import_importlib(module_name):
     """Import a module from importlib both w/ and w/o _frozen_importlib."""
     fresh = ('importlib',) if '.' in module_name else ()
@@ -38,6 +75,9 @@ if sys.platform not in ('win32', 'cygwin'):
     if not os.path.exists(changed_name):
         CASE_INSENSITIVE_FS = False
 
+_, source_importlib = import_importlib('importlib')
+__import__ = staticmethod(builtins.__import__), staticmethod(source_importlib.__import__)
+
 
 def case_insensitive_tests(test):
     """Class decorator that nullifies tests requiring a case-insensitive
@@ -53,7 +93,7 @@ def submodule(parent, name, pkg_dir, content=''):
     return '{}.{}'.format(parent, name), path
 
 
-@contextmanager
+@contextlib.contextmanager
 def uncache(*names):
     """Uncache a module from sys.modules.
 
@@ -79,7 +119,7 @@ def uncache(*names):
                 pass
 
 
-@contextmanager
+@contextlib.contextmanager
 def temp_module(name, content='', *, pkg=False):
     conflicts = [n for n in sys.modules if n.partition('.')[0] == name]
     with support.temp_cwd(None) as cwd:
@@ -103,7 +143,7 @@ def temp_module(name, content='', *, pkg=False):
                 yield location
 
 
-@contextmanager
+@contextlib.contextmanager
 def import_state(**kwargs):
     """Context manager to manage the various importers and stored state in the
     sys module.
@@ -198,6 +238,7 @@ class mock_modules(_ImporterMock):
                     raise
             return self.modules[fullname]
 
+
 class mock_spec(_ImporterMock):
 
     """Importer mock using PEP 451 APIs."""
@@ -223,3 +264,99 @@ class mock_spec(_ImporterMock):
             self.module_code[module.__spec__.name]()
         except KeyError:
             pass
+
+
+def writes_bytecode_files(fxn):
+    """Decorator to protect sys.dont_write_bytecode from mutation and to skip
+    tests that require it to be set to False."""
+    if sys.dont_write_bytecode:
+        return lambda *args, **kwargs: None
+    @functools.wraps(fxn)
+    def wrapper(*args, **kwargs):
+        original = sys.dont_write_bytecode
+        sys.dont_write_bytecode = False
+        try:
+            to_return = fxn(*args, **kwargs)
+        finally:
+            sys.dont_write_bytecode = original
+        return to_return
+    return wrapper
+
+
+def ensure_bytecode_path(bytecode_path):
+    """Ensure that the __pycache__ directory for PEP 3147 pyc file exists.
+
+    :param bytecode_path: File system path to PEP 3147 pyc file.
+    """
+    try:
+        os.mkdir(os.path.dirname(bytecode_path))
+    except OSError as error:
+        if error.errno != errno.EEXIST:
+            raise
+
+
+@contextlib.contextmanager
+def create_modules(*names):
+    """Temporarily create each named module with an attribute (named 'attr')
+    that contains the name passed into the context manager that caused the
+    creation of the module.
+
+    All files are created in a temporary directory returned by
+    tempfile.mkdtemp(). This directory is inserted at the beginning of
+    sys.path. When the context manager exits all created files (source and
+    bytecode) are explicitly deleted.
+
+    No magic is performed when creating packages! This means that if you create
+    a module within a package you must also create the package's __init__ as
+    well.
+
+    """
+    source = 'attr = {0!r}'
+    created_paths = []
+    mapping = {}
+    state_manager = None
+    uncache_manager = None
+    try:
+        temp_dir = tempfile.mkdtemp()
+        mapping['.root'] = temp_dir
+        import_names = set()
+        for name in names:
+            if not name.endswith('__init__'):
+                import_name = name
+            else:
+                import_name = name[:-len('.__init__')]
+            import_names.add(import_name)
+            if import_name in sys.modules:
+                del sys.modules[import_name]
+            name_parts = name.split('.')
+            file_path = temp_dir
+            for directory in name_parts[:-1]:
+                file_path = os.path.join(file_path, directory)
+                if not os.path.exists(file_path):
+                    os.mkdir(file_path)
+                    created_paths.append(file_path)
+            file_path = os.path.join(file_path, name_parts[-1] + '.py')
+            with open(file_path, 'w') as file:
+                file.write(source.format(name))
+            created_paths.append(file_path)
+            mapping[name] = file_path
+        uncache_manager = uncache(*import_names)
+        uncache_manager.__enter__()
+        state_manager = import_state(path=[temp_dir])
+        state_manager.__enter__()
+        yield mapping
+    finally:
+        if state_manager is not None:
+            state_manager.__exit__(None, None, None)
+        if uncache_manager is not None:
+            uncache_manager.__exit__(None, None, None)
+        support.rmtree(temp_dir)
+
+
+def mock_path_hook(*entries, importer):
+    """A mock sys.path_hooks entry."""
+    def hook(entry):
+        if entry not in entries:
+            raise ImportError
+        return importer
+    return hook