bpo-43105: importlib now loads dynamic modules using full paths

Author: zooba

Doc/library/os.path.rst

@@ -459,6 +459,11 @@ the :mod:`glob` module.)
    .. versionchanged:: 3.6
       Accepts a :term:`path-like object`.
 
+   .. versionchanged:: 3.10
+      On Windows, now relies on an operating system API. This has changed the
+      results for some invalid UNC paths, but reliably supports paths with
+      special prefixes such as ``\\?\``.
+
 
 .. function:: splitext(path)
 

Doc/whatsnew/3.10.rst

@@ -793,6 +793,11 @@ Added :data:`~os.O_EVTONLY`, :data:`~os.O_FSYNC`, :data:`~os.O_SYMLINK`
 and :data:`~os.O_NOFOLLOW_ANY` for macOS.
 (Contributed by Dong-hee Na in :issue:`43106`.)
 
+Changed :func:`os.path.splitdrive()` on Windows to rely on a native API.
+This has improved support for special prefixes, but may have changed the
+results for some invalid UNC paths.
+(Contributed by Steve Dower in :issue:`43105`.)
+
 pathlib
 -------
 

Lib/importlib/_bootstrap_external.py

@@ -45,6 +45,7 @@
 # Assumption made in _path_join()
 assert all(len(sep) == 1 for sep in path_separators)
 path_sep = path_separators[0]
+path_sep_tuple = tuple(path_separators)
 path_separators = ''.join(path_separators)
 _pathseps_with_colon = {f':{s}' for s in path_separators}
 
@@ -91,22 +92,48 @@ def _unpack_uint16(data):
     return int.from_bytes(data, 'little')
 
 
-def _path_join(*path_parts):
-    """Replacement for os.path.join()."""
-    return path_sep.join([part.rstrip(path_separators)
-                          for part in path_parts if part])
+if _MS_WINDOWS:
+    def _path_join(*path_parts):
+        """Replacement for os.path.join()."""
+        if not path_parts:
+            return ""
+        if len(path_parts) == 1:
+            return path_parts[0]
+        root = ""
+        path = []
+        for new_root, tail in map(_os._path_splitroot, path_parts):
+            if new_root.startswith(path_sep_tuple) or new_root.endswith(path_sep_tuple):
+                root = new_root.rstrip(path_separators) or root
+                path = [path_sep + tail]
+            elif new_root.endswith(':'):
+                if root.casefold() != new_root.casefold():
+                    # Drive relative paths have to be resolved by the OS, so we reset the
+                    # tail but do not add a path_sep prefix.
+                    root = new_root
+                    path = [tail]
+                else:
+                    path.append(tail)
+            else:
+                root = new_root or root
+                path.append(tail)
+        if len(path) == 1 and path[0] in path_sep_tuple:
+            # Avoid losing the root's trailing separator when joining with nothing
+            return root + path_sep
+        return root + path_sep.join(p.rstrip(path_separators) for p in path if p)
+
+else:
+    def _path_join(*path_parts):
+        """Replacement for os.path.join()."""
+        return path_sep.join([part.rstrip(path_separators)
+                              for part in path_parts if part])
 
 
 def _path_split(path):
     """Replacement for os.path.split()."""
-    if len(path_separators) == 1:
-        front, _, tail = path.rpartition(path_sep)
-        return front, tail
-    for x in reversed(path):
-        if x in path_separators:
-            front, tail = path.rsplit(x, maxsplit=1)
-            return front, tail
-    return '', path
+    i = max(path.rfind(p) for p in path_separators)
+    if i < 0:
+        return '', path
+    return path[:i], path[i + 1:]
 
 
 def _path_stat(path):
@@ -140,13 +167,18 @@ def _path_isdir(path):
     return _path_is_mode_type(path, 0o040000)
 
 
-def _path_isabs(path):
-    """Replacement for os.path.isabs.
+if _MS_WINDOWS:
+    def _path_isabs(path):
+        """Replacement for os.path.isabs."""
+        if not path:
+            return False
+        root = _os._path_splitroot(path)[0].replace('/', '\\')
+        return len(root) > 1 and (root.startswith('\\\\') or root.endswith('\\'))
 
-    Considers a Windows drive-relative path (no drive, but starts with slash) to
-    still be "absolute".
-    """
-    return path.startswith(path_separators) or path[1:3] in _pathseps_with_colon
+else:
+    def _path_isabs(path):
+        """Replacement for os.path.isabs."""
+        return path.startswith(path_separators)
 
 
 def _write_atomic(path, data, mode=0o666):
@@ -707,6 +739,11 @@ def spec_from_file_location(name, location=None, *, loader=None,
                 pass
     else:
         location = _os.fspath(location)
+        if not _path_isabs(location):
+            try:
+                location = _path_join(_os.getcwd(), location)
+            except OSError:
+                pass
 
     # If the location is on the filesystem, but doesn't actually exist,
     # we could return None here, indicating that the location is not
@@ -1451,6 +1488,8 @@ def __init__(self, path, *loader_details):
         self._loaders = loaders
         # Base (directory) path
         self.path = path or '.'
+        if not _path_isabs(self.path):
+            self.path = _path_join(_os.getcwd(), self.path)
         self._path_mtime = -1
         self._path_cache = set()
         self._relaxed_path_cache = set()
@@ -1516,7 +1555,10 @@ def find_spec(self, fullname, target=None):
                 is_namespace = _path_isdir(base_path)
         # Check for a file w/ a proper suffix exists.
         for suffix, loader_class in self._loaders:
-            full_path = _path_join(self.path, tail_module + suffix)
+            try:
+                full_path = _path_join(self.path, tail_module + suffix)
+            except ValueError:
+                return None
             _bootstrap._verbose_message('trying {}', full_path, verbosity=2)
             if cache_module + suffix in cache:
                 if _path_isfile(full_path):

Lib/ntpath.py

@@ -58,8 +58,21 @@ def normcase(s):
 # volume), or if a pathname after the volume-letter-and-colon or UNC-resource
 # starts with a slash or backslash.
 
+try:
+    from nt import _path_splitroot
+except ImportError:
+    _path_splitroot = None
+
+
 def isabs(s):
     """Test whether a path is absolute"""
+    if _path_splitroot:
+        try:
+            r = _path_splitroot(os.fsdecode(s))[0].replace(altsep, sep)
+        except ValueError:
+            pass
+        else:
+            return r.startswith(sep + sep) or r.endswith(sep)
     s = os.fspath(s)
     # Paths beginning with \\?\ are always absolute, but do not
     # necessarily contain a drive.
@@ -141,17 +154,31 @@ def splitdrive(p):
 
     """
     p = os.fspath(p)
-    if len(p) >= 2:
-        if isinstance(p, bytes):
-            sep = b'\\'
-            altsep = b'/'
-            colon = b':'
+    if isinstance(p, bytes):
+        sep = b'\\'
+        altsep = b'/'
+        colon = b':'
+    else:
+        sep = '\\'
+        altsep = '/'
+        colon = ':'
+
+    if _path_splitroot:
+        try:
+            d = _path_splitroot(p)[0].rstrip('\\/')
+        except ValueError:
+            pass
         else:
-            sep = '\\'
-            altsep = '/'
-            colon = ':'
+            if not d and p[1:2] == colon:
+                # Special case for invalid drive letters (such as a wildcard in glob.escape)
+                d = p[:2]
+            if isinstance(p, bytes):
+                d = os.fsencode(d)
+            return p[:len(d)], p[len(d):]
+
+    if len(p) >= 2:
         normp = p.replace(altsep, sep)
-        if (normp[0:2] == sep*2) and (normp[2:3] != sep):
+        if (normp[0:2] == sep*2):
             # is a UNC path:
             # vvvvvvvvvvvvvvvvvvvv drive letter or UNC path
             # \\machine\mountpoint\directory\etc\...
@@ -160,10 +187,11 @@ def splitdrive(p):
             if index == -1:
                 return p[:0], p
             index2 = normp.find(sep, index + 1)
-            # a UNC path can't have two slashes in a row
-            # (after the initial two)
+            # a UNC path shouldn't have two slashes in a row
+            # (after the initial two), but to be consistent with
+            # the native function, we split before them.
             if index2 == index + 1:
-                return p[:0], p
+                return p[:index], p[index:]
             if index2 == -1:
                 index2 = len(p)
             return p[:index2], p[index2:]
@@ -262,15 +290,25 @@ def lexists(path):
 def ismount(path):
     """Test whether a path is a mount point (a drive root, the root of a
     share, or a mounted volume)"""
+    try_fallback = True
+    if _path_splitroot:
+        try:
+            root, tail = _path_splitroot(os.fsdecode(path))
+        except ValueError:
+            pass
+        else:
+            try_fallback = False
+            if not tail:
+                root = root.replace(altsep, sep)
+                return root.startswith(sep + sep) or root.endswith(sep)
+
+    if try_fallback:
+        root, rest = splitdrive(os.fsdecode(path))
+        if root:
+            return rest in ("/", "\\") or root.startswith(("\\\\", "//"))
+
     path = os.fspath(path)
     seps = _get_bothseps(path)
-    path = abspath(path)
-    root, rest = splitdrive(path)
-    if root and root[0] in seps:
-        return (not rest) or (rest in seps)
-    if rest in seps:
-        return True
-
     if _getvolumepathname:
         return path.rstrip(seps) == _getvolumepathname(path).rstrip(seps)
     else:
@@ -505,7 +543,7 @@ def _abspath_fallback(path):
     """
 
     path = os.fspath(path)
-    if not isabs(path):
+    if not splitdrive(path)[0]:
         if isinstance(path, bytes):
             cwd = os.getcwdb()
         else:
@@ -671,9 +709,9 @@ def realpath(path):
         return path
 
 
-# Win9x family and earlier have no Unicode filename support.
-supports_unicode_filenames = (hasattr(sys, "getwindowsversion") and
-                              sys.getwindowsversion()[3] >= 2)
+# All supported platforms have Unicode filenames
+supports_unicode_filenames = True
+
 
 def relpath(path, start=None):
     """Return a relative version of a path"""

Lib/test/test_import/__init__.py

@@ -906,15 +906,15 @@ def test_missing_source_legacy(self):
         m = __import__(TESTFN)
         try:
             self.assertEqual(m.__file__,
-                             os.path.join(os.curdir, os.path.relpath(pyc_file)))
+                             os.path.join(os.getcwd(), os.curdir, os.path.relpath(pyc_file)))
         finally:
             os.remove(pyc_file)
 
     def test___cached__(self):
         # Modules now also have an __cached__ that points to the pyc file.
         m = __import__(TESTFN)
         pyc_file = importlib.util.cache_from_source(TESTFN + '.py')
-        self.assertEqual(m.__cached__, os.path.join(os.curdir, pyc_file))
+        self.assertEqual(m.__cached__, os.path.join(os.getcwd(), os.curdir, pyc_file))
 
     @skip_if_dont_write_bytecode
     def test___cached___legacy_pyc(self):
@@ -930,7 +930,7 @@ def test___cached___legacy_pyc(self):
         importlib.invalidate_caches()
         m = __import__(TESTFN)
         self.assertEqual(m.__cached__,
-                         os.path.join(os.curdir, os.path.relpath(pyc_file)))
+                         os.path.join(os.getcwd(), os.curdir, os.path.relpath(pyc_file)))
 
     @skip_if_dont_write_bytecode
     def test_package___cached__(self):
@@ -950,10 +950,10 @@ def cleanup():
         m = __import__('pep3147.foo')
         init_pyc = importlib.util.cache_from_source(
             os.path.join('pep3147', '__init__.py'))
-        self.assertEqual(m.__cached__, os.path.join(os.curdir, init_pyc))
+        self.assertEqual(m.__cached__, os.path.join(os.getcwd(), os.curdir, init_pyc))
         foo_pyc = importlib.util.cache_from_source(os.path.join('pep3147', 'foo.py'))
         self.assertEqual(sys.modules['pep3147.foo'].__cached__,
-                         os.path.join(os.curdir, foo_pyc))
+                         os.path.join(os.getcwd(), os.curdir, foo_pyc))
 
     def test_package___cached___from_pyc(self):
         # Like test___cached__ but ensuring __cached__ when imported from a
@@ -977,10 +977,10 @@ def cleanup():
         m = __import__('pep3147.foo')
         init_pyc = importlib.util.cache_from_source(
             os.path.join('pep3147', '__init__.py'))
-        self.assertEqual(m.__cached__, os.path.join(os.curdir, init_pyc))
+        self.assertEqual(m.__cached__, os.path.join(os.getcwd(), os.curdir, init_pyc))
         foo_pyc = importlib.util.cache_from_source(os.path.join('pep3147', 'foo.py'))
         self.assertEqual(sys.modules['pep3147.foo'].__cached__,
-                         os.path.join(os.curdir, foo_pyc))
+                         os.path.join(os.getcwd(), os.curdir, foo_pyc))
 
     def test_recompute_pyc_same_second(self):
         # Even when the source file doesn't change timestamp, a change in

Lib/test/test_importlib/test_spec.py

@@ -506,7 +506,7 @@ class FactoryTests:
 
     def setUp(self):
         self.name = 'spam'
-        self.path = 'spam.py'
+        self.path = os.path.abspath('spam.py')
         self.cached = self.util.cache_from_source(self.path)
         self.loader = TestLoader()
         self.fileloader = TestLoader(self.path)
@@ -645,7 +645,7 @@ def test_spec_from_loader_is_package_true_with_fileloader(self):
         self.assertEqual(spec.loader, self.fileloader)
         self.assertEqual(spec.origin, self.path)
         self.assertIs(spec.loader_state, None)
-        self.assertEqual(spec.submodule_search_locations, [''])
+        self.assertEqual(spec.submodule_search_locations, [os.getcwd()])
         self.assertEqual(spec.cached, self.cached)
         self.assertTrue(spec.has_location)
 
@@ -744,7 +744,7 @@ def test_spec_from_file_location_smsl_empty(self):
         self.assertEqual(spec.loader, self.fileloader)
         self.assertEqual(spec.origin, self.path)
         self.assertIs(spec.loader_state, None)
-        self.assertEqual(spec.submodule_search_locations, [''])
+        self.assertEqual(spec.submodule_search_locations, [os.getcwd()])
         self.assertEqual(spec.cached, self.cached)
         self.assertTrue(spec.has_location)
 
@@ -769,7 +769,7 @@ def test_spec_from_file_location_smsl_default(self):
         self.assertEqual(spec.loader, self.pkgloader)
         self.assertEqual(spec.origin, self.path)
         self.assertIs(spec.loader_state, None)
-        self.assertEqual(spec.submodule_search_locations, [''])
+        self.assertEqual(spec.submodule_search_locations, [os.getcwd()])
         self.assertEqual(spec.cached, self.cached)
         self.assertTrue(spec.has_location)
 
@@ -817,6 +817,17 @@ def is_package(self, name):
         self.assertEqual(spec.cached, self.cached)
         self.assertTrue(spec.has_location)
 
+    def test_spec_from_file_location_relative_path(self):
+        spec = self.util.spec_from_file_location(self.name,
+            os.path.basename(self.path), loader=self.fileloader)
+
+        self.assertEqual(spec.name, self.name)
+        self.assertEqual(spec.loader, self.fileloader)
+        self.assertEqual(spec.origin, self.path)
+        self.assertIs(spec.loader_state, None)
+        self.assertIs(spec.submodule_search_locations, None)
+        self.assertEqual(spec.cached, self.cached)
+        self.assertTrue(spec.has_location)
 
 (Frozen_FactoryTests,
  Source_FactoryTests