Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/_pytest/pathlib.py: 19%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3from collections.abc import Callable
4from collections.abc import Iterable
5from collections.abc import Iterator
6import contextlib
7from enum import Enum
8from errno import EBADF
9from errno import ELOOP
10from errno import ENOENT
11from errno import ENOTDIR
12import fnmatch
13from functools import partial
14from importlib.machinery import ModuleSpec
15from importlib.machinery import PathFinder
16import importlib.util
17import itertools
18import os
19from os.path import expanduser
20from os.path import expandvars
21from os.path import isabs
22from os.path import sep
23from pathlib import Path
24from pathlib import PurePath
25from posixpath import sep as posix_sep
26import shutil
27import stat
28import sys
29import types
30from types import ModuleType
31from typing import Any
32from typing import TypeVar
33import uuid
34import warnings
36from _pytest.compat import assert_never
37from _pytest.outcomes import skip
38from _pytest.warning_types import PytestWarning
41if sys.version_info < (3, 11):
42 from importlib._bootstrap_external import _NamespaceLoader as NamespaceLoader
43else:
44 from importlib.machinery import NamespaceLoader
46LOCK_TIMEOUT = 60 * 60 * 24 * 3
48_AnyPurePath = TypeVar("_AnyPurePath", bound=PurePath)
50# The following function, variables and comments were
51# copied from cpython 3.9 Lib/pathlib.py file.
53# EBADF - guard against macOS `stat` throwing EBADF
54_IGNORED_ERRORS = (ENOENT, ENOTDIR, EBADF, ELOOP)
56_IGNORED_WINERRORS = (
57 21, # ERROR_NOT_READY - drive exists but is not accessible
58 1921, # ERROR_CANT_RESOLVE_FILENAME - fix for broken symlink pointing to itself
59)
62def _ignore_error(exception: Exception) -> bool:
63 return (
64 getattr(exception, "errno", None) in _IGNORED_ERRORS
65 or getattr(exception, "winerror", None) in _IGNORED_WINERRORS
66 )
69def get_lock_path(path: _AnyPurePath) -> _AnyPurePath:
70 return path.joinpath(".lock")
73def _chmod_rwx(p: str) -> bool:
74 """Grant owner sufficient permissions for deletion.
76 Directories get ``S_IRWXU`` (read+write+exec for traversal).
77 Regular files get ``S_IRUSR | S_IWUSR`` only, to avoid making
78 non-executable files executable as a side effect.
80 Returns True if permissions were actually changed, False if they were
81 already sufficient or couldn't be changed.
82 """
83 try:
84 old_mode = os.stat(p).st_mode
85 except OSError:
86 # Path may have been removed concurrently, or be inaccessible.
87 return False
88 perm_mode = stat.S_IMODE(old_mode)
89 bits = stat.S_IRWXU if stat.S_ISDIR(old_mode) else stat.S_IRUSR | stat.S_IWUSR
90 new_mode = perm_mode | bits
91 if perm_mode == new_mode:
92 return False
93 try:
94 os.chmod(p, new_mode)
95 except OSError:
96 return False
97 return True
100def on_rm_rf_error(
101 func: Callable[..., Any] | None,
102 path: str,
103 excinfo: BaseException
104 | tuple[type[BaseException], BaseException, types.TracebackType | None],
105 *,
106 start_path: Path,
107) -> bool:
108 """Handle known read-only errors during rmtree.
110 The returned value is used only by our own tests.
111 """
112 if isinstance(excinfo, BaseException):
113 exc = excinfo
114 else:
115 exc = excinfo[1]
117 # Another process removed the file in the middle of the "rm_rf" (xdist for example).
118 # More context: https://github.com/pytest-dev/pytest/issues/5974#issuecomment-543799018
119 if isinstance(exc, FileNotFoundError):
120 return False
122 if not isinstance(exc, PermissionError):
123 warnings.warn(
124 PytestWarning(f"(rm_rf) error removing {path}\n{type(exc)}: {exc}")
125 )
126 return False
128 p = Path(path)
130 if func in (os.open, os.scandir):
131 # Directory traversal failed (e.g. missing S_IXUSR). Fix permissions
132 # on the path and its parent (bounded by start_path), then remove it
133 # ourselves since rmtree skips entries after the error handler returns.
134 # See: https://github.com/pytest-dev/pytest/issues/7940
135 parent_changed = False
136 parent = p.parent
137 # Never chmod outside the tree rooted at start_path.
138 if parent not in (p, start_path):
139 parent_changed = _chmod_rwx(str(parent))
140 path_changed = _chmod_rwx(str(p))
141 if not (parent_changed or path_changed):
142 return False
143 if p.is_dir():
144 rm_rf(p)
145 else:
146 try:
147 os.unlink(str(p))
148 except OSError:
149 return False
150 return True
152 if func not in (os.rmdir, os.remove, os.unlink):
153 warnings.warn(
154 PytestWarning(
155 f"(rm_rf) unknown function {func} when removing {path}:\n{type(exc)}: {exc}"
156 )
157 )
158 return False
160 # Chmod + retry.
161 # For files, we need to recursively go upwards in the directories to
162 # ensure they all are also accessible and writable.
163 if p.is_file():
164 for parent in p.parents: # pragma: no branch
165 _chmod_rwx(str(parent))
166 # Stop when we reach the original path passed to rm_rf.
167 if parent == start_path:
168 break
169 _chmod_rwx(str(path))
171 func(path)
172 return True
175def ensure_extended_length_path(path: Path) -> Path:
176 """Get the extended-length version of a path (Windows).
178 On Windows, by default, the maximum length of a path (MAX_PATH) is 260
179 characters, and operations on paths longer than that fail. But it is possible
180 to overcome this by converting the path to "extended-length" form before
181 performing the operation:
182 https://docs.microsoft.com/en-us/windows/win32/fileio/naming-a-file#maximum-path-length-limitation
184 On Windows, this function returns the extended-length absolute version of path.
185 On other platforms it returns path unchanged.
186 """
187 if sys.platform.startswith("win32"):
188 path = path.resolve()
189 path = Path(get_extended_length_path_str(str(path)))
190 return path
193def get_extended_length_path_str(path: str) -> str:
194 """Convert a path to a Windows extended length path."""
195 long_path_prefix = "\\\\?\\"
196 unc_long_path_prefix = "\\\\?\\UNC\\"
197 if path.startswith((long_path_prefix, unc_long_path_prefix)):
198 return path
199 # UNC
200 if path.startswith("\\\\"):
201 return unc_long_path_prefix + path[2:]
202 return long_path_prefix + path
205def rm_rf(path: Path) -> None:
206 """Remove the path contents recursively, even if some elements
207 are read-only."""
208 path = ensure_extended_length_path(path)
209 onerror = partial(on_rm_rf_error, start_path=path)
210 if sys.version_info >= (3, 12):
211 shutil.rmtree(str(path), onexc=onerror)
212 else:
213 shutil.rmtree(str(path), onerror=onerror)
216def find_prefixed(root: Path, prefix: str) -> Iterator[os.DirEntry[str]]:
217 """Find all elements in root that begin with the prefix, case-insensitive."""
218 l_prefix = prefix.lower()
219 for x in os.scandir(root):
220 if x.name.lower().startswith(l_prefix):
221 yield x
224def extract_suffixes(iter: Iterable[os.DirEntry[str]], prefix: str) -> Iterator[str]:
225 """Return the parts of the paths following the prefix.
227 :param iter: Iterator over path names.
228 :param prefix: Expected prefix of the path names.
229 """
230 p_len = len(prefix)
231 for entry in iter:
232 yield entry.name[p_len:]
235def find_suffixes(root: Path, prefix: str) -> Iterator[str]:
236 """Combine find_prefixes and extract_suffixes."""
237 return extract_suffixes(find_prefixed(root, prefix), prefix)
240def parse_num(maybe_num: str) -> int:
241 """Parse number path suffixes, returns -1 on error."""
242 try:
243 return int(maybe_num)
244 except ValueError:
245 return -1
248def _force_symlink(root: Path, target: str | PurePath, link_to: str | Path) -> None:
249 """Helper to create the current symlink.
251 It's full of race conditions that are reasonably OK to ignore
252 for the context of best effort linking to the latest test run.
254 The presumption being that in case of much parallelism
255 the inaccuracy is going to be acceptable.
256 """
257 current_symlink = root.joinpath(target)
258 try:
259 current_symlink.unlink()
260 except OSError:
261 pass
262 try:
263 current_symlink.symlink_to(link_to)
264 except Exception:
265 pass
268def make_numbered_dir(root: Path, prefix: str, mode: int = 0o700) -> Path:
269 """Create a directory with an increased number as suffix for the given prefix."""
270 for i in range(10):
271 # try up to 10 times to create the directory
272 max_existing = max(map(parse_num, find_suffixes(root, prefix)), default=-1)
273 new_number = max_existing + 1
274 new_path = root.joinpath(f"{prefix}{new_number}")
275 try:
276 new_path.mkdir(mode=mode)
277 except Exception:
278 pass
279 else:
280 _force_symlink(root, prefix + "current", new_path)
281 return new_path
282 else:
283 raise OSError(
284 "could not create numbered dir with prefix "
285 f"{prefix} in {root} after 10 tries"
286 )
289def create_cleanup_lock(p: Path) -> Path:
290 """Create a lock to prevent premature directory cleanup."""
291 lock_path = get_lock_path(p)
292 try:
293 fd = os.open(str(lock_path), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644)
294 except FileExistsError as e:
295 raise OSError(f"cannot create lockfile in {p}") from e
296 else:
297 pid = os.getpid()
298 spid = str(pid).encode()
299 os.write(fd, spid)
300 os.close(fd)
301 if not lock_path.is_file():
302 raise OSError("lock path got renamed after successful creation")
303 return lock_path
306def register_cleanup_lock_removal(lock_path: Path, register: Any) -> Any:
307 """Register a cleanup function for removing a lock."""
308 pid = os.getpid()
310 def cleanup_on_exit(lock_path: Path = lock_path, original_pid: int = pid) -> None:
311 current_pid = os.getpid()
312 if current_pid != original_pid:
313 # fork
314 return
315 try:
316 lock_path.unlink()
317 except OSError:
318 pass
320 return register(cleanup_on_exit)
323def maybe_delete_a_numbered_dir(path: Path) -> None:
324 """Remove a numbered directory if its lock can be obtained and it does
325 not seem to be in use."""
326 path = ensure_extended_length_path(path)
327 lock_path = None
328 try:
329 lock_path = create_cleanup_lock(path)
330 parent = path.parent
332 garbage = parent.joinpath(f"garbage-{uuid.uuid4()}")
333 path.rename(garbage)
334 rm_rf(garbage)
335 except OSError:
336 # known races:
337 # * other process did a cleanup at the same time
338 # * deletable directory was found
339 # * process cwd (Windows)
340 return
341 finally:
342 # If we created the lock, ensure we remove it even if we failed
343 # to properly remove the numbered dir.
344 if lock_path is not None:
345 try:
346 lock_path.unlink()
347 except OSError:
348 pass
351def ensure_deletable(path: Path, consider_lock_dead_if_created_before: float) -> bool:
352 """Check if `path` is deletable based on whether the lock file is expired."""
353 if path.is_symlink():
354 return False
355 lock = get_lock_path(path)
356 try:
357 if not lock.is_file():
358 return True
359 except OSError:
360 # we might not have access to the lock file at all, in this case assume
361 # we don't have access to the entire directory (#7491).
362 return False
363 try:
364 lock_time = lock.stat().st_mtime
365 except Exception:
366 return False
367 else:
368 if lock_time < consider_lock_dead_if_created_before:
369 # We want to ignore any errors while trying to remove the lock such as:
370 # - PermissionDenied, like the file permissions have changed since the lock creation;
371 # - FileNotFoundError, in case another pytest process got here first;
372 # and any other cause of failure.
373 with contextlib.suppress(OSError):
374 lock.unlink()
375 return True
376 return False
379def try_cleanup(path: Path, consider_lock_dead_if_created_before: float) -> None:
380 """Try to cleanup a directory if we can ensure it's deletable."""
381 if ensure_deletable(path, consider_lock_dead_if_created_before):
382 maybe_delete_a_numbered_dir(path)
385def cleanup_candidates(root: Path, prefix: str, keep: int) -> Iterator[Path]:
386 """List candidates for numbered directories to be removed - follows py.path."""
387 max_existing = max(map(parse_num, find_suffixes(root, prefix)), default=-1)
388 max_delete = max_existing - keep
389 entries = find_prefixed(root, prefix)
390 entries, entries2 = itertools.tee(entries)
391 numbers = map(parse_num, extract_suffixes(entries2, prefix))
392 for entry, number in zip(entries, numbers, strict=True):
393 if number <= max_delete:
394 yield Path(entry)
397def cleanup_dead_symlinks(root: Path) -> None:
398 for left_dir in root.iterdir():
399 if left_dir.is_symlink():
400 if not left_dir.resolve().exists():
401 left_dir.unlink()
404def cleanup_numbered_dir(
405 root: Path, prefix: str, keep: int, consider_lock_dead_if_created_before: float
406) -> None:
407 """Cleanup for lock driven numbered directories."""
408 if not root.exists():
409 return
410 for path in cleanup_candidates(root, prefix, keep):
411 try_cleanup(path, consider_lock_dead_if_created_before)
412 for path in root.glob("garbage-*"):
413 try_cleanup(path, consider_lock_dead_if_created_before)
415 cleanup_dead_symlinks(root)
418def make_numbered_dir_with_cleanup(
419 *,
420 root: Path,
421 prefix: str,
422 mode: int,
423 keep: int,
424 lock_timeout: float,
425 register: Any,
426) -> Path:
427 """Create a numbered dir and register its cleanup.
429 Similar to make_numbered_dir, but also maintains a lock file indicating that
430 the directory is currently in use, and registers the cleanup of the lock and
431 of stale numbered directories.
433 :param keep:
434 The number of sessions to retain the directory.
435 :param lock_timeout:
436 In case of a crash, the lock remains "stuck". The timeout is a time
437 limit after which the lock is considered stale and can be removed.
438 :param register:
439 Called as register(cleanup_func, params...). Should schedule to call
440 passed cleanup functions on session finish.
441 """
442 e = None
443 for i in range(10):
444 try:
445 p = make_numbered_dir(root, prefix, mode)
446 # Only lock the current dir when keep is not 0
447 if keep != 0:
448 lock_path = create_cleanup_lock(p)
449 register_cleanup_lock_removal(lock_path, register)
450 except Exception as exc:
451 e = exc
452 else:
453 consider_lock_dead_if_created_before = p.stat().st_mtime - lock_timeout
454 # Register a cleanup for program exit
455 register(
456 cleanup_numbered_dir,
457 root,
458 prefix,
459 keep,
460 consider_lock_dead_if_created_before,
461 )
462 return p
463 assert e is not None
464 raise e
467def resolve_from_str(input: str, rootpath: Path) -> Path:
468 input = expanduser(input)
469 input = expandvars(input)
470 if isabs(input):
471 return Path(input)
472 else:
473 return rootpath.joinpath(input)
476def fnmatch_ex(pattern: str, path: str | os.PathLike[str]) -> bool:
477 """A port of FNMatcher from py.path.common which works with PurePath() instances.
479 The difference between this algorithm and PurePath.match() is that the
480 latter matches "**" glob expressions for each part of the path, while
481 this algorithm uses the whole path instead.
483 For example:
484 "tests/foo/bar/doc/test_foo.py" matches pattern "tests/**/doc/test*.py"
485 with this algorithm, but not with PurePath.match().
487 This algorithm was ported to keep backward-compatibility with existing
488 settings which assume paths match according this logic.
490 References:
491 * https://bugs.python.org/issue29249
492 * https://bugs.python.org/issue34731
493 """
494 path = PurePath(path)
495 iswin32 = sys.platform.startswith("win")
497 if iswin32 and sep not in pattern and posix_sep in pattern:
498 # Running on Windows, the pattern has no Windows path separators,
499 # and the pattern has one or more Posix path separators. Replace
500 # the Posix path separators with the Windows path separator.
501 pattern = pattern.replace(posix_sep, sep)
503 if sep not in pattern:
504 name = path.name
505 else:
506 name = str(path)
507 if path.is_absolute() and not os.path.isabs(pattern):
508 pattern = f"*{os.sep}{pattern}"
509 return fnmatch.fnmatch(name, pattern)
512def parts(s: str) -> set[str]:
513 parts = s.split(sep)
514 return {sep.join(parts[: i + 1]) or sep for i in range(len(parts))}
517def symlink_or_skip(
518 src: os.PathLike[str] | str,
519 dst: os.PathLike[str] | str,
520 **kwargs: Any,
521) -> None:
522 """Make a symlink, or skip the test in case symlinks are not supported."""
523 try:
524 os.symlink(src, dst, **kwargs)
525 except OSError as e:
526 skip(f"symlinks not supported: {e}")
529class ImportMode(Enum):
530 """Possible values for `mode` parameter of `import_path`."""
532 prepend = "prepend"
533 append = "append"
534 importlib = "importlib"
537class ImportPathMismatchError(ImportError):
538 """Raised on import_path() if there is a mismatch of __file__'s.
540 This can happen when `import_path` is called multiple times with different filenames that has
541 the same basename but reside in packages
542 (for example "/tests1/test_foo.py" and "/tests2/test_foo.py").
543 """
546def import_path(
547 path: str | os.PathLike[str],
548 *,
549 mode: str | ImportMode = ImportMode.prepend,
550 root: Path,
551 consider_namespace_packages: bool,
552) -> ModuleType:
553 """
554 Import and return a module from the given path, which can be a file (a module) or
555 a directory (a package).
557 :param path:
558 Path to the file to import.
560 :param mode:
561 Controls the underlying import mechanism that will be used:
563 * ImportMode.prepend: the directory containing the module (or package, taking
564 `__init__.py` files into account) will be put at the *start* of `sys.path` before
565 being imported with `importlib.import_module`.
567 * ImportMode.append: same as `prepend`, but the directory will be appended
568 to the end of `sys.path`, if not already in `sys.path`.
570 * ImportMode.importlib: uses more fine control mechanisms provided by `importlib`
571 to import the module, which avoids having to muck with `sys.path` at all. It effectively
572 allows having same-named test modules in different places.
574 :param root:
575 Used as an anchor when mode == ImportMode.importlib to obtain
576 a unique name for the module being imported so it can safely be stored
577 into ``sys.modules``.
579 :param consider_namespace_packages:
580 If True, consider namespace packages when resolving module names.
582 :raises ImportPathMismatchError:
583 If after importing the given `path` and the module `__file__`
584 are different. Only raised in `prepend` and `append` modes.
585 """
586 path = Path(path)
587 mode = ImportMode(mode)
589 if not path.exists():
590 raise ImportError(path)
592 if mode is ImportMode.importlib:
593 # Try to import this module using the standard import mechanisms, but
594 # without touching sys.path.
595 try:
596 _, module_name = resolve_pkg_root_and_module_name(
597 path, consider_namespace_packages=consider_namespace_packages
598 )
599 except CouldNotResolvePathError:
600 pass
601 else:
602 # If the given module name is already in sys.modules, do not import it again.
603 with contextlib.suppress(KeyError):
604 return sys.modules[module_name]
606 mod = _import_module_using_spec(module_name, path, insert_modules=False)
607 if mod is not None:
608 return mod
610 # Could not import the module with the current sys.path, so we fall back
611 # to importing the file as a single module, not being a part of a package.
612 module_name = module_name_from_path(path, root)
613 with contextlib.suppress(KeyError):
614 return sys.modules[module_name]
616 mod = _import_module_using_spec(module_name, path, insert_modules=True)
617 if mod is None:
618 raise ImportError(f"Can't find module {module_name} at location {path}")
619 return mod
621 try:
622 pkg_root, module_name = resolve_pkg_root_and_module_name(
623 path, consider_namespace_packages=consider_namespace_packages
624 )
625 except CouldNotResolvePathError:
626 pkg_root, module_name = path.parent, path.stem
628 # Change sys.path permanently: restoring it at the end of this function would cause surprising
629 # problems because of delayed imports: for example, a conftest.py file imported by this function
630 # might have local imports, which would fail at runtime if we restored sys.path.
631 if mode is ImportMode.append:
632 if str(pkg_root) not in sys.path:
633 sys.path.append(str(pkg_root))
634 elif mode is ImportMode.prepend:
635 if str(pkg_root) != sys.path[0]:
636 sys.path.insert(0, str(pkg_root))
637 else:
638 assert_never(mode)
640 importlib.import_module(module_name)
642 mod = sys.modules[module_name]
643 if path.name == "__init__.py":
644 return mod
646 ignore = os.environ.get("PY_IGNORE_IMPORTMISMATCH", "")
647 if ignore != "1":
648 module_file = mod.__file__
649 if module_file is None:
650 raise ImportPathMismatchError(module_name, module_file, path)
652 if module_file.endswith((".pyc", ".pyo")):
653 module_file = module_file[:-1]
654 module_file = module_file.removesuffix(os.sep + "__init__.py")
656 try:
657 is_same = _is_same(str(path), module_file)
658 except FileNotFoundError:
659 is_same = False
661 if not is_same:
662 raise ImportPathMismatchError(module_name, module_file, path)
664 return mod
667def _import_module_using_spec(
668 module_name: str, module_path: Path, *, insert_modules: bool
669) -> ModuleType | None:
670 """
671 Tries to import a module by its canonical name, path, and its parent location.
673 :param module_name:
674 The expected module name, will become the key of `sys.modules`.
676 :param module_path:
677 The file path of the module, for example `/foo/bar/test_demo.py`.
678 If module is a package, pass the path to the `__init__.py` of the package.
679 If module is a namespace package, pass directory path.
681 :param insert_modules:
682 If True, will call `insert_missing_modules` to create empty intermediate modules
683 with made-up module names (when importing test files not reachable from `sys.path`).
685 Example 1 of parent_module_*:
687 module_name: "a.b.c.demo"
688 module_path: Path("a/b/c/demo.py")
689 if "a.b.c" is package ("a/b/c/__init__.py" exists), then
690 parent_module_name: "a.b.c"
691 parent_module_path: Path("a/b/c/__init__.py")
692 else:
693 parent_module_name: "a.b.c"
694 parent_module_path: Path("a/b/c")
696 Example 2 of parent_module_*:
698 module_name: "a.b.c"
699 module_path: Path("a/b/c/__init__.py")
700 if "a.b" is package ("a/b/__init__.py" exists), then
701 parent_module_name: "a.b"
702 parent_module_path: Path("a/b/__init__.py")
703 else:
704 parent_module_name: "a.b"
705 parent_module_path: Path("a/b/")
706 """
707 # Attempt to import the parent module, seems is our responsibility:
708 # https://github.com/python/cpython/blob/73906d5c908c1e0b73c5436faeff7d93698fc074/Lib/importlib/_bootstrap.py#L1308-L1311
709 parent_module_name, _, name = module_name.rpartition(".")
710 parent_module: ModuleType | None = None
711 if parent_module_name:
712 parent_module = sys.modules.get(parent_module_name)
713 # If the parent_module lacks the `__path__` attribute, AttributeError when finding a submodule's spec,
714 # requiring re-import according to the path.
715 need_reimport = not hasattr(parent_module, "__path__")
716 if parent_module is None or need_reimport:
717 # Get parent_location based on location, get parent_path based on path.
718 if module_path.name == "__init__.py":
719 # If the current module is in a package,
720 # need to leave the package first and then enter the parent module.
721 parent_module_path = module_path.parent.parent
722 else:
723 parent_module_path = module_path.parent
725 if (parent_module_path / "__init__.py").is_file():
726 # If the parent module is a package, loading by __init__.py file.
727 parent_module_path = parent_module_path / "__init__.py"
729 parent_module = _import_module_using_spec(
730 parent_module_name,
731 parent_module_path,
732 insert_modules=insert_modules,
733 )
735 # Checking with sys.meta_path first in case one of its hooks can import this module,
736 # such as our own assertion-rewrite hook.
737 find_spec_path = [str(module_path.parent)]
738 for meta_importer in sys.meta_path:
739 spec = meta_importer.find_spec(module_name, find_spec_path)
741 if spec_matches_module_path(spec, module_path):
742 break
743 else:
744 loader = None
745 if module_path.is_dir():
746 # The `spec_from_file_location` matches a loader based on the file extension by default.
747 # For a namespace package, need to manually specify a loader.
748 loader = NamespaceLoader(name, module_path, PathFinder()) # type: ignore[arg-type]
750 spec = importlib.util.spec_from_file_location(
751 module_name, str(module_path), loader=loader
752 )
754 if spec_matches_module_path(spec, module_path):
755 assert spec is not None
756 # Find spec and import this module.
757 mod = importlib.util.module_from_spec(spec)
758 sys.modules[module_name] = mod
759 spec.loader.exec_module(mod) # type: ignore[union-attr]
761 # Set this module as an attribute of the parent module (#12194).
762 if parent_module is not None:
763 setattr(parent_module, name, mod)
765 if insert_modules:
766 insert_missing_modules(sys.modules, module_name)
767 return mod
769 return None
772def spec_matches_module_path(module_spec: ModuleSpec | None, module_path: Path) -> bool:
773 """Return true if the given ModuleSpec can be used to import the given module path."""
774 if module_spec is None:
775 return False
777 if module_spec.origin:
778 return Path(module_spec.origin) == module_path
780 # Compare the path with the `module_spec.submodule_Search_Locations` in case
781 # the module is part of a namespace package.
782 # https://docs.python.org/3/library/importlib.html#importlib.machinery.ModuleSpec.submodule_search_locations
783 if module_spec.submodule_search_locations: # can be None.
784 for path in module_spec.submodule_search_locations:
785 if Path(path) == module_path:
786 return True
788 return False
791# Implement a special _is_same function on Windows which returns True if the two filenames
792# compare equal, to circumvent os.path.samefile returning False for mounts in UNC (#7678).
793if sys.platform.startswith("win"):
795 def _is_same(f1: str, f2: str) -> bool:
796 return Path(f1) == Path(f2) or os.path.samefile(f1, f2)
798else:
800 def _is_same(f1: str, f2: str) -> bool:
801 return os.path.samefile(f1, f2)
804def module_name_from_path(path: Path, root: Path) -> str:
805 """
806 Return a dotted module name based on the given path, anchored on root.
808 For example: path="projects/src/tests/test_foo.py" and root="/projects", the
809 resulting module name will be "src.tests.test_foo".
810 """
811 path = path.with_suffix("")
812 try:
813 relative_path = path.relative_to(root)
814 except ValueError:
815 # If we can't get a relative path to root, use the full path, except
816 # for the first part ("d:\\" or "/" depending on the platform, for example).
817 path_parts = path.parts[1:]
818 else:
819 # Use the parts for the relative path to the root path.
820 path_parts = relative_path.parts
822 # Module name for packages do not contain the __init__ file, unless
823 # the `__init__.py` file is at the root.
824 if len(path_parts) >= 2 and path_parts[-1] == "__init__":
825 path_parts = path_parts[:-1]
827 # Module names cannot contain ".", normalize them to "_". This prevents
828 # a directory having a "." in the name (".env.310" for example) causing extra intermediate modules.
829 # Also, important to replace "." at the start of paths, as those are considered relative imports.
830 path_parts = tuple(x.replace(".", "_") for x in path_parts)
832 return ".".join(path_parts)
835def insert_missing_modules(modules: dict[str, ModuleType], module_name: str) -> None:
836 """
837 Used by ``import_path`` to create intermediate modules when using mode=importlib.
839 When we want to import a module as "src.tests.test_foo" for example, we need
840 to create empty modules "src" and "src.tests" after inserting "src.tests.test_foo",
841 otherwise "src.tests.test_foo" is not importable by ``__import__``.
842 """
843 module_parts = module_name.split(".")
844 while module_name:
845 parent_module_name, _, child_name = module_name.rpartition(".")
846 if parent_module_name:
847 parent_module = modules.get(parent_module_name)
848 if parent_module is None:
849 try:
850 # If sys.meta_path is empty, calling import_module will issue
851 # a warning and raise ModuleNotFoundError. To avoid the
852 # warning, we check sys.meta_path explicitly and raise the error
853 # ourselves to fall back to creating a dummy module.
854 if not sys.meta_path:
855 raise ModuleNotFoundError
856 parent_module = importlib.import_module(parent_module_name)
857 except ModuleNotFoundError:
858 parent_module = ModuleType(
859 module_name,
860 doc="Empty module created by pytest's importmode=importlib.",
861 )
862 modules[parent_module_name] = parent_module
864 # Add child attribute to the parent that can reference the child
865 # modules.
866 if not hasattr(parent_module, child_name):
867 setattr(parent_module, child_name, modules[module_name])
869 module_parts.pop(-1)
870 module_name = ".".join(module_parts)
873def resolve_package_path(path: Path) -> Path | None:
874 """Return the Python package path by looking for the last
875 directory upwards which still contains an __init__.py.
877 Returns None if it cannot be determined.
878 """
879 result = None
880 for parent in itertools.chain((path,), path.parents):
881 if parent.is_dir():
882 if not (parent / "__init__.py").is_file():
883 break
884 if not parent.name.isidentifier():
885 break
886 result = parent
887 return result
890def resolve_pkg_root_and_module_name(
891 path: Path, *, consider_namespace_packages: bool = False
892) -> tuple[Path, str]:
893 """
894 Return the path to the directory of the root package that contains the
895 given Python file, and its module name:
897 src/
898 app/
899 __init__.py
900 core/
901 __init__.py
902 models.py
904 Passing the full path to `models.py` will yield Path("src") and "app.core.models".
906 If consider_namespace_packages is True, then we additionally check upwards in the hierarchy
907 for namespace packages:
909 https://packaging.python.org/en/latest/guides/packaging-namespace-packages
911 Raises CouldNotResolvePathError if the given path does not belong to a package (missing any __init__.py files).
912 """
913 pkg_root: Path | None = None
914 pkg_path = resolve_package_path(path)
915 if pkg_path is not None:
916 pkg_root = pkg_path.parent
917 if consider_namespace_packages:
918 start = pkg_root if pkg_root is not None else path.parent
919 for candidate in (start, *start.parents):
920 module_name = compute_module_name(candidate, path)
921 if module_name and is_importable(module_name, path):
922 # Point the pkg_root to the root of the namespace package.
923 pkg_root = candidate
924 break
926 if pkg_root is not None:
927 module_name = compute_module_name(pkg_root, path)
928 if module_name:
929 return pkg_root, module_name
931 raise CouldNotResolvePathError(f"Could not resolve for {path}")
934def is_importable(module_name: str, module_path: Path) -> bool:
935 """
936 Return if the given module path could be imported normally by Python, akin to the user
937 entering the REPL and importing the corresponding module name directly, and corresponds
938 to the module_path specified.
940 :param module_name:
941 Full module name that we want to check if is importable.
942 For example, "app.models".
944 :param module_path:
945 Full path to the python module/package we want to check if is importable.
946 For example, "/projects/src/app/models.py".
947 """
948 try:
949 # Note this is different from what we do in ``_import_module_using_spec``, where we explicitly search through
950 # sys.meta_path to be able to pass the path of the module that we want to import (``meta_importer.find_spec``).
951 # Using importlib.util.find_spec() is different, it gives the same results as trying to import
952 # the module normally in the REPL.
953 spec = importlib.util.find_spec(module_name)
954 except (ImportError, ValueError, ImportWarning):
955 return False
956 else:
957 return spec_matches_module_path(spec, module_path)
960def compute_module_name(root: Path, module_path: Path) -> str | None:
961 """Compute a module name based on a path and a root anchor."""
962 try:
963 path_without_suffix = module_path.with_suffix("")
964 except ValueError:
965 # Empty paths (such as Path.cwd()) might break meta_path hooks (like our own assertion rewriter).
966 return None
968 try:
969 relative = path_without_suffix.relative_to(root)
970 except ValueError: # pragma: no cover
971 return None
972 names = list(relative.parts)
973 if not names:
974 return None
975 if names[-1] == "__init__":
976 names.pop()
977 return ".".join(names)
980class CouldNotResolvePathError(Exception):
981 """Custom exception raised by resolve_pkg_root_and_module_name."""
984def scandir(
985 path: str | os.PathLike[str],
986 sort_key: Callable[[os.DirEntry[str]], object] = lambda entry: entry.name,
987) -> list[os.DirEntry[str]]:
988 """Scan a directory recursively, in breadth-first order.
990 The returned entries are sorted according to the given key.
991 The default is to sort by name.
992 If the directory does not exist, return an empty list.
993 """
994 entries = []
995 # Attempt to create a scandir iterator for the given path.
996 try:
997 scandir_iter = os.scandir(path)
998 except FileNotFoundError:
999 # If the directory does not exist, return an empty list.
1000 return []
1001 # Use the scandir iterator in a context manager to ensure it is properly closed.
1002 with scandir_iter as s:
1003 for entry in s:
1004 try:
1005 entry.is_file()
1006 except OSError as err:
1007 if _ignore_error(err):
1008 continue
1009 # Reraise non-ignorable errors to avoid hiding issues.
1010 raise
1011 entries.append(entry)
1012 entries.sort(key=sort_key) # type: ignore[arg-type]
1013 return entries
1016def visit(
1017 path: str | os.PathLike[str], recurse: Callable[[os.DirEntry[str]], bool]
1018) -> Iterator[os.DirEntry[str]]:
1019 """Walk a directory recursively, in breadth-first order.
1021 The `recurse` predicate determines whether a directory is recursed.
1023 Entries at each directory level are sorted.
1024 """
1025 entries = scandir(path)
1026 yield from entries
1027 for entry in entries:
1028 if entry.is_dir() and recurse(entry):
1029 yield from visit(entry.path, recurse)
1032def absolutepath(path: str | os.PathLike[str]) -> Path:
1033 """Convert a path to an absolute path using os.path.abspath.
1035 Prefer this over Path.resolve() (see #6523).
1036 Prefer this over Path.absolute() (not public, doesn't normalize).
1037 """
1038 return Path(os.path.abspath(path))
1041def commonpath(path1: Path, path2: Path) -> Path | None:
1042 """Return the common part shared with the other path, or None if there is
1043 no common part.
1045 If one path is relative and one is absolute, returns None.
1046 """
1047 try:
1048 return Path(os.path.commonpath((str(path1), str(path2))))
1049 except ValueError:
1050 return None
1053def bestrelpath(directory: Path, dest: Path) -> str:
1054 """Return a string which is a relative path from directory to dest such
1055 that directory/bestrelpath == dest.
1057 The paths must be either both absolute or both relative.
1059 If no such path can be determined, returns dest.
1060 """
1061 assert isinstance(directory, Path)
1062 assert isinstance(dest, Path)
1063 if dest == directory:
1064 return os.curdir
1065 # Find the longest common directory.
1066 base = commonpath(directory, dest)
1067 # Can be the case on Windows for two absolute paths on different drives.
1068 # Can be the case for two relative paths without common prefix.
1069 # Can be the case for a relative path and an absolute path.
1070 if not base:
1071 return str(dest)
1072 reldirectory = directory.relative_to(base)
1073 reldest = dest.relative_to(base)
1074 return os.path.join(
1075 # Back from directory to base.
1076 *([os.pardir] * len(reldirectory.parts)),
1077 # Forward from base to dest.
1078 *reldest.parts,
1079 )
1082def safe_exists(p: Path) -> bool:
1083 """Like Path.exists(), but account for input arguments that might be too long (#11394)."""
1084 try:
1085 return p.exists()
1086 except (ValueError, OSError):
1087 # ValueError: stat: path too long for Windows
1088 # OSError: [WinError 123] The filename, directory name, or volume label syntax is incorrect
1089 return False
1092def samefile_nofollow(p1: Path, p2: Path) -> bool:
1093 """Test whether two paths reference the same actual file or directory.
1095 Unlike Path.samefile(), does not resolve symlinks.
1096 """
1097 s1, s2 = p1.lstat(), p2.lstat()
1098 # On Windows st_ino is the file ID, which file systems are free to not support,
1099 # in which case it is 0 for every file -- WinFsp mounts such as sshfs-win are one
1100 # example. os.path.samestat() would then consider any two files on the volume to
1101 # be the same (python/cpython#78116), so treat a zero file ID as "unknown" and
1102 # leave the caller with plain path comparison (#14864).
1103 if s1.st_ino == 0 or s2.st_ino == 0:
1104 return False
1105 return os.path.samestat(s1, s2)