diff --git a/Lib/site.py b/Lib/site.py index b6d840fb2..72bdf0b9f 100644 --- a/Lib/site.py +++ b/Lib/site.py @@ -18,55 +18,26 @@ sys.prefix and sys.exec_prefix are set to that directory and it is also checked for site-packages (sys.base_prefix and sys.base_exec_prefix will always be the "real" prefixes of the Python installation). If "pyvenv.cfg" (a bootstrap configuration file) contains -the key "include-system-site-packages" set to anything other than "false" -(case-insensitive), the system-level prefixes will still also be -searched for site-packages; otherwise they won't. +the key "include-system-site-packages" set to "true" (case-insensitive), +the system-level prefixes will still also be searched for site-packages; +otherwise they won't. -All of the resulting site-specific directories, if they exist, are -appended to sys.path, and also inspected for path configuration -files. +Two kinds of configuration files are processed in each site-packages +directory: -A path configuration file is a file whose name has the form -.pth; its contents are additional directories (one per line) -to be added to sys.path. Non-existing directories (or -non-directories) are never added to sys.path; no directory is added to -sys.path more than once. Blank lines and lines beginning with -'#' are skipped. Lines starting with 'import' are executed. +- .pth files extend sys.path with additional directories (one per + line). Lines starting with "import" are deprecated (see PEP 829). -For example, suppose sys.prefix and sys.exec_prefix are set to -/usr/local and there is a directory /usr/local/lib/python2.5/site-packages -with three subdirectories, foo, bar and spam, and two path -configuration files, foo.pth and bar.pth. Assume foo.pth contains the -following: +- .start files specify startup entry points using the pkg.mod:callable + syntax. These are resolved via pkgutil.resolve_name() and called with no + arguments. - # foo package configuration - foo - bar - bletch +When called from main(), all .pth path extensions are applied before any +.start entry points are executed, ensuring that paths are available before +startup code runs. -and bar.pth contains: - - # bar package configuration - bar - -Then the following directories are added to sys.path, in this order: - - /usr/local/lib/python2.5/site-packages/bar - /usr/local/lib/python2.5/site-packages/foo - -Note that bletch is omitted because it doesn't exist; bar precedes foo -because bar.pth comes alphabetically before foo.pth; and spam is -omitted because it is not mentioned in either path configuration file. - -The readline module is also automatically configured to enable -completion for systems that support it. This can be overridden in -sitecustomize, usercustomize or PYTHONSTARTUP. Starting Python in -isolated mode (-I) disables automatic readline configuration. - -After these operations, an attempt is made to import a module -named sitecustomize, which can perform arbitrary additional -site-specific customizations. If this import fails with an -ImportError exception, it is silently ignored. +See the documentation for the site module for full details: +https://docs.python.org/3/library/site.html """ import sys @@ -77,6 +48,11 @@ import _io as io import stat import errno +lazy import locale +lazy import pkgutil +lazy import traceback +lazy import warnings + # Prefixes for site-packages; add additional prefixes like /usr/local here PREFIXES = [sys.prefix, sys.exec_prefix] # Enable per user site-packages directory @@ -90,17 +66,34 @@ USER_SITE = None USER_BASE = None -def _trace(message): +def _trace(message, exc=None): if sys.flags.verbose: - print(message, file=sys.stderr) + _print_error(message, exc) + + +def _print_error(message, exc=None): + """Print an error message to stderr, optionally with a formatted traceback.""" + print(message, file=sys.stderr) + if exc is not None: + for record in traceback.format_exception(exc): + for line in record.splitlines(): + print(' ' + line, file=sys.stderr) def _warn(*args, **kwargs): - import warnings - warnings.warn(*args, **kwargs) +def _warn_future_us(message, remove): + # Don't call warnings._deprecated() directly because we're lazily importing warnings and don't + # want to have to trigger an eager import if it's not necessary. Startup time matters a lot + # here and warnings isn't cheap! This inlines the check from + # warnings._py_warnings._deprecated(). + _version = sys.version_info + if (_version[:2] > remove) or (_version[:2] == remove and _version[3] != "alpha"): + warnings._deprecated(message, remove=remove) + + def makepath(*paths): dir = os.path.join(*paths) try: @@ -111,7 +104,7 @@ def makepath(*paths): def abs_paths(): - """Set all module __file__ and __cached__ attributes to an absolute path""" + """Set __file__ to an absolute path.""" for m in set(sys.modules.values()): loader_module = None try: @@ -127,10 +120,6 @@ def abs_paths(): m.__file__ = os.path.abspath(m.__file__) except (AttributeError, OSError, TypeError): pass - try: - m.__cached__ = os.path.abspath(m.__cached__) - except (AttributeError, OSError, TypeError): - pass def removeduppaths(): @@ -165,95 +154,461 @@ def _init_pathinfo(): return d -def addpackage(sitedir, name, known_paths): - """Process a .pth file within the site-packages directory: - For each line in the file, either combine it with sitedir to a path - and add that to known_paths, or execute it if it starts with 'import '. +# PEP 829 implementation notes. +# +# Startup information (.pth and .start file information) can be processed in +# implicit or explicit batches. Implicit batches are self-contained +# site.addsitedir() calls: they create a per-call StartupState, populate it +# from the site directory's .pth and .start files, run process() on it, and +# then throw the state away. +# +# main() needs different semantics: it accumulates state across multiple +# StartupState.addsitedir() calls (user-site plus all global site-packages) so +# that every sys.path extension is visible *before* any startup code (.start +# entry points and .pth import lines) runs. Callers can opt into the same +# behavior by creating a StartupState directly and calling its addsitedir(), +# addusersitepackages(), and addsitepackages() methods, then invoking +# process() once at the end of the batch. +# +# Here's the CRITICAL reentrancy invariant: recursive site.addsitedir() calls +# reached from a .start entry point or an exec'd .pth import line must not +# mutate the StartupState currently being processed. Reentrant calls reach +# the module-level site.addsitedir() shim, which always builds a fresh +# per-call state. + + +def _read_pthstart_file(sitedir, name, suffix): + """Parse a .start or .pth file and return (lines, filename). + + On success, ``lines`` is a (possibly empty) list of the file's lines. + On failure (file missing, hidden, unreadable, or .start with bad + encoding), ``lines`` is ``None`` so callers can distinguish a + successfully-read empty file from one that could not be read. """ + filename = os.path.join(sitedir, name) + _trace(f"Reading startup configuration file: {filename}") + + try: + st = os.lstat(filename) + except OSError as exc: + _trace(f"Cannot stat {filename!r}", exc) + return None, filename + + if ((getattr(st, 'st_flags', 0) & stat.UF_HIDDEN) or + (getattr(st, 'st_file_attributes', 0) & stat.FILE_ATTRIBUTE_HIDDEN)): + _trace(f"Skipping hidden {suffix} file: {filename!r}") + return None, filename + + _trace(f"Processing {suffix} file: {filename!r}") + try: + with io.open_code(filename) as f: + raw_content = f.read() + except OSError as exc: + _trace(f"Cannot read {filename!r}", exc) + return None, filename + + try: + # Accept BOM markers in .start and .pth files as we do in source files + # (Windows PowerShell 5.1 makes it hard to emit UTF-8 files without a BOM). + content = raw_content.decode("utf-8-sig") + except UnicodeDecodeError: + _trace(f"Cannot read {filename!r} as UTF-8.") + # For .pth files only, and then only until Python 3.20, fall back to + # locale encoding for backward compatibility. + _warn_future_us( + ".pth files decoded to locale encoding as a fallback", + remove=(3, 20) + ) + if suffix == ".pth": + content = raw_content.decode(locale.getencoding()) + _trace(f"Using fallback encoding {locale.getencoding()!r}") + else: + return None, filename + + return content.splitlines(), filename + + +class StartupState: + """Per-batch accumulator for .pth and .start file processing. + + A StartupState collects sys.path extensions, deprecated .pth import lines, + and .start entry points read from one or more site-packages directories. + Calling process() applies them in PEP 829 order: paths are added to + sys.path first, then import lines from .pth files (skipping any with a + matching .start), then entry points from .start files. + + State lives entirely on the instance; there is no module-level pending + state. This is what makes the module reentrancy-safe: a site.addsitedir() + call reached recursively from an exec'd import line or a .start entry + point operates on a different StartupState than the one being processed by + the outer call. + + The internal data is intentionally private. The lower-level write + methods (_record_sitedir(), _read_pth_file(), _read_start_file()) are + private to the site module; the public surface is addsitedir(), + addusersitepackages(), addsitepackages(), and process(). + """ + __slots__ = ( + '_known_paths', + '_processed_sitedirs', + '_path_entries', + '_importexecs', + '_entrypoints', + ) + + def __init__(self, known_paths=None): + """Create an independent startup state. + + *known_paths* is a set of case-normalized paths already present + on sys.path, used to avoid duplicate path entries. When None + (the default), it is initialized from the current sys.path. + + A caller-supplied set is stored by reference and mutated in place + as new paths are recorded; pass a fresh set per StartupState if + isolation across instances is required. + """ + self._known_paths = ( + _init_pathinfo() + if known_paths is None + else known_paths) + self._processed_sitedirs = set() + # The sys.path append ledger. This is a list of 2-tuples of the form + # (pthfile, path) where `pthfile` is the .pth file which is extending + # the path, and `path` is the directory to add to sys.path. Note that + # to preserve the interleaving semantics (i.e. .pth file paths are + # added after the sitedir in which the .pth file is found), `path` + # could be a sitedir, in which case `pthfile` will always be None. + self._path_entries = [] + # Both dicts map "" -> list + # of items collected from that file. Mapping by filename lets us + # cross-reference a .pth and its matching .start (PEP 829 import + # suppression rule) and lets _print_error report the source file + # when an entry fails. + self._importexecs = {} + self._entrypoints = {} + + def addsitedir(self, sitedir): + """Add a site directory and accumulate its .pth and .start startup data. + + Read the .pth and .start files in *sitedir* and record their + sys.path extensions, deprecated .pth import lines, and .start entry + points on this state. The recorded data is not applied until + process() is called. + + Typically used to batch multiple site directories before a single + process() call, so that every sys.path extension is visible before + any startup code runs. Reentrant calls reached from a .start entry + point or an exec'd .pth import line must not mutate the state + currently being processed; for those cases, use site.addsitedir() + instead, which always creates a fresh per-call state. + """ + self._addsitedir(sitedir, process_known_sitedirs=True) + + def addusersitepackages(self): + """Add the per-user site-packages directory, if enabled. + + The user site directory is added only when user site-packages are + enabled and the directory exists. Its startup data is accumulated + for later processing by process(). + """ + _trace("Processing user site-packages") + user_site = getusersitepackages() + if ENABLE_USER_SITE and os.path.isdir(user_site): + self.addsitedir(user_site) + + def addsitepackages(self, prefixes=None): + """Add global site-packages directories, if they exist. + + Site-packages directories are computed from *prefixes*, or from the + global PREFIXES when *prefixes* is None. Each directory's startup + data is accumulated for later processing by process(). + """ + _trace("Processing global site-packages") + for sitedir in getsitepackages(prefixes): + if os.path.isdir(sitedir): + self.addsitedir(sitedir) + + def _addsitedir(self, sitedir, *, process_known_sitedirs): + """Internal addsitedir() implementation with full dedup control. + + The public addsitedir() always uses process_known_sitedirs=True + (gh-149819 semantics). The module-level legacy known_paths shim + uses process_known_sitedirs=False to preserve 3.14 idempotency + (gh-75723). + """ + sitedir = self._record_sitedir( + sitedir, process_known_sitedirs=process_known_sitedirs) + if sitedir is None: + return + try: + names = os.listdir(sitedir) + except OSError: + return + + # The following phases are defined by PEP 829. + # Phases 1-3: Read .pth files, accumulating paths and import lines. + pth_names = sorted( + name for name in names + if name.endswith(".pth") and not name.startswith(".") + ) + for name in pth_names: + self._read_pth_file(sitedir, name) + + # Phases 6-7: Discover .start files and accumulate their entry points. + # Import lines from .pth files with a matching .start file are + # discarded at flush time by _exec_imports(). + start_names = sorted( + name for name in names + if name.endswith(".start") and not name.startswith(".") + ) + for name in start_names: + self._read_start_file(sitedir, name) + + def _record_sitedir(self, sitedir, *, process_known_sitedirs=True): + sitedir, sitedircase = makepath(sitedir) + # Have we already processed this sitedir? + if sitedircase in self._processed_sitedirs: + return None + # In legacy known_paths mode, a known sitedir means its startup files + # were already processed by an earlier addsitedir() call, so skip it + # to preserve idempotency (gh-75723). In explicit StartupState mode, + # known_paths only tracks sys.path entries; a sitedir may already be + # on sys.path (for example from $PYTHONPATH, gh-149819) but still need + # its .pth and .start files processed once. The separate + # _processed_sitedirs set is what lets explicit batches distinguish + # "already on sys.path" from "startup files already read". + if not process_known_sitedirs and sitedircase in self._known_paths: + return None + # Record that we've processed this sitedir. + self._processed_sitedirs.add(sitedircase) + if sitedircase not in self._known_paths: + self._known_paths.add(sitedircase) + # Add the sitedir to the sys.path extension ledger. There is no + # .pth file to record. + self._path_entries.append((None, sitedir)) + return sitedir + + def _read_pth_file(self, sitedir, name): + """Parse a .pth file, accumulating sys.path extensions and import lines. + + Errors on individual lines do not abort processing of the rest of + the file (PEP 829). Per-batch deduplication is done against + self._known_paths: any path already in it is skipped, and newly + accepted paths are added to it so that subsequent .pth files in + the same batch don't add them more than once. + """ + lines, filename = _read_pthstart_file(sitedir, name, ".pth") + if lines is None: + return + + for n, line in enumerate(lines, 1): + line = line.strip() + if not line or line.startswith("#"): + continue + + # In Python 3.18 and 3.19, `import` lines are silently + # ignored. In Python 3.20 and beyond, issue a warning when + # `import` lines in .pth files are detected. + if line.startswith(("import ", "import\t")): + _warn_future_us( + "import lines in .pth files are silently ignored", + remove=(3, 18), + ) + _warn_future_us( + "import lines in .pth files are noisily ignored", + remove=(3, 20), + ) + self._importexecs.setdefault(filename, []).append(line) + continue + + try: + dir_, dircase = makepath(sitedir, line) + except Exception as exc: + _trace(f"Error in {filename!r}, line {n:d}: {line!r}", exc) + continue + + # PEP 829 dedup: skip paths already seen in this batch. + if dircase in self._known_paths: + _trace( + f"In {filename!r}, line {n:d}: " + f"skipping duplicate sys.path entry: {dir_}" + ) + else: + # Add this directory to the sys.path extension ledger, while + # also recording the .pth file it was found in. + self._path_entries.append((filename, dir_)) + self._known_paths.add(dircase) + + def _read_start_file(self, sitedir, name): + """Parse a .start file for a list of entry point strings.""" + lines, filename = _read_pthstart_file(sitedir, name, ".start") + if lines is None: + return + + # PEP 829: the *presence* of a matching .start file disables `import` + # line processing in the matched .pth file, regardless of whether this + # .start file contains any entry points. Register the filename as a + # key now so an empty (or comment-only) .start file still suppresses. + entrypoints = self._entrypoints.setdefault(filename, []) + + for n, line in enumerate(lines, 1): + line = line.strip() + if not line or line.startswith("#"): + continue + # Syntax validation is deferred to entry point execution + # time, where pkgutil.resolve_name(strict=True) enforces the + # pkg.mod:callable form. + entrypoints.append(line) + + def process(self): + """Apply accumulated state in PEP 829 order. + + Phase order matters: all .pth path extensions are applied to + sys.path *before* any import line or .start entry point runs, so + that an entry point may live in a module reachable only via a + .pth-extended path. + """ + self._extend_syspath() + self._exec_imports() + self._execute_start_entrypoints() + + def _extend_syspath(self): + # Duplicate path-extension specifications have already been filtered + # out upstream across .pth files within this batch (via known_paths), + # and ledger entries are already abspath/normpath'd. .pth-derived + # entries (filename is not None) are existence-checked and skipped + # with an error if missing. Sitedir entries (filename is None) are + # appended unconditionally: legacy addsitedir() added the sitedir to + # sys.path before attempting to list it, so an unreadable or + # non-existent sitedir still landed on sys.path. Deferring the + # append to here preserves that contract. + for filename, dir_ in self._path_entries: + # As a backstop, known_paths may not have been seeded from sys.path + # (callers can pass an empty set), and multiple StartupState + # instances against the same sys.path don't share state, so always + # do a final anti-duplication check. + if dir_ in sys.path: + continue + if filename is None or os.path.exists(dir_): + if filename is not None: + _trace(f"Extending sys.path with {dir_} from {filename}") + sys.path.append(dir_) + else: + _print_error( + f"In {filename}: {dir_} does not exist; " + f"skipping sys.path append" + ) + + def _exec_imports(self): + # For each `import` line we've seen in a .pth file, exec() it in + # order, unless the .pth has a matching .start file in this same + # batch. In that case, PEP 829 says the import lines are + # suppressed in favor of the .start's entry points. + for filename, imports in self._importexecs.items(): + # Inject 'sitedir' local variable in the current frame for + # compatibility with Python 3.14. Especially, "-nspkg.pth" files + # generated by setuptools use: sys._getframe(1).f_locals['sitedir']. + sitedir = os.path.dirname(filename) + + # Inject 'fullname' local variable in the current frame for + # compatibility with Python 3.14, where the path of the ".pth" + # file being processed was available to import lines under that + # name. + fullname = filename + + # Given "/path/to/foo.pth", check whether "/path/to/foo.start" was + # registered in this same batch. + name, dot, pth = filename.rpartition(".") + assert dot == "." and pth == "pth", ( + f"Bad startup filename: {filename}" + ) + if f"{name}.start" in self._entrypoints: + _trace( + f"import lines in {filename} are suppressed " + f"due to matching {name}.start file." + ) + continue + + _trace( + f"import lines in {filename} are deprecated, " + f"use entry points in a {name}.start file instead." + ) + for line in imports: + try: + _trace(f"Exec'ing from {filename}: {line}") + exec(line) + except Exception as exc: + _print_error( + f"Error in import line from {filename}: {line}", + exc, + ) + + def _execute_start_entrypoints(self): + # Resolve each entry point string to a callable via + # pkgutil.resolve_name(strict=True), which both validates the + # required pkg.mod:callable form and performs the import in one + # step, then call it with no arguments. + for filename, entrypoints in self._entrypoints.items(): + for entrypoint in entrypoints: + try: + _trace( + f"Executing entry point: {entrypoint} from {filename}" + ) + callable_ = pkgutil.resolve_name(entrypoint, strict=True) + except ValueError as exc: + _print_error( + f"Invalid entry point syntax in {filename}: " + f"{entrypoint!r}", + exc, + ) + except Exception as exc: + _print_error( + f"Error resolving entry point {entrypoint} " + f"from {filename}", + exc, + ) + else: + try: + callable_() + except Exception as exc: + _print_error( + f"Error in entry point {entrypoint} from {filename}", + exc, + ) + + +def addpackage(sitedir, name, known_paths): + """Process a .pth file within the site-packages directory.""" if known_paths is None: known_paths = _init_pathinfo() reset = True else: reset = False - fullname = os.path.join(sitedir, name) - try: - st = os.lstat(fullname) - except OSError: - return - if ((getattr(st, 'st_flags', 0) & stat.UF_HIDDEN) or - (getattr(st, 'st_file_attributes', 0) & stat.FILE_ATTRIBUTE_HIDDEN)): - _trace(f"Skipping hidden .pth file: {fullname!r}") - return - _trace(f"Processing .pth file: {fullname!r}") - try: - with io.open_code(fullname) as f: - pth_content = f.read() - except OSError: - return - try: - # Accept BOM markers in .pth files as we do in source files - # (Windows PowerShell 5.1 makes it hard to emit UTF-8 files without a BOM) - pth_content = pth_content.decode("utf-8-sig") - except UnicodeDecodeError: - # Fallback to locale encoding for backward compatibility. - # We will deprecate this fallback in the future. - import locale - pth_content = pth_content.decode(locale.getencoding()) - _trace(f"Cannot read {fullname!r} as UTF-8. " - f"Using fallback encoding {locale.getencoding()!r}") + state = StartupState(known_paths) + state._read_pth_file(sitedir, name) + state.process() - for n, line in enumerate(pth_content.splitlines(), 1): - if line.startswith("#"): - continue - if line.strip() == "": - continue - try: - if line.startswith(("import ", "import\t")): - exec(line) - continue - line = line.rstrip() - dir, dircase = makepath(sitedir, line) - if dircase not in known_paths and os.path.exists(dir): - sys.path.append(dir) - known_paths.add(dircase) - except Exception as exc: - print(f"Error processing line {n:d} of {fullname}:\n", - file=sys.stderr) - import traceback - for record in traceback.format_exception(exc): - for line in record.splitlines(): - print(' '+line, file=sys.stderr) - print("\nRemainder of file ignored", file=sys.stderr) - break - if reset: - known_paths = None - return known_paths + return None if reset else known_paths def addsitedir(sitedir, known_paths=None): - """Add 'sitedir' argument to sys.path if missing and handle .pth files in - 'sitedir'""" + """Add a site directory and process its startup files. + + For batched processing across multiple site directories, build a + StartupState explicitly and call StartupState.addsitedir() on it; that + defers .pth/.start processing until a single StartupState.process() call. + """ _trace(f"Adding directory: {sitedir!r}") if known_paths is None: - known_paths = _init_pathinfo() - reset = True + state = StartupState(_init_pathinfo()) + state.addsitedir(sitedir) else: - reset = False - sitedir, sitedircase = makepath(sitedir) - if not sitedircase in known_paths: - sys.path.append(sitedir) # Add path component - known_paths.add(sitedircase) - try: - names = os.listdir(sitedir) - except OSError: - return - names = [name for name in names - if name.endswith(".pth") and not name.startswith(".")] - for name in sorted(names): - addpackage(sitedir, name, known_paths) - if reset: - known_paths = None + # Preserve gh-75723 idempotency for legacy known_paths mode: a + # sitedir already present in known_paths is skipped, not reprocessed. + state = StartupState(known_paths) + state._addsitedir(sitedir, process_known_sitedirs=False) + state.process() return known_paths @@ -333,8 +688,7 @@ def _get_path(userbase): if sys.platform == 'darwin' and sys._framework: return f'{userbase}/lib/{implementation_lower}/site-packages' - # XXX: RUSTPYTHON - return f'{userbase}/lib/rustpython{version[0]}.{version[1]}{abi_thread}/site-packages' + return f'{userbase}/lib/{implementation_lower}{version[0]}.{version[1]}{abi_thread}/site-packages' def getuserbase(): @@ -367,21 +721,20 @@ def getusersitepackages(): return USER_SITE + def addusersitepackages(known_paths): - """Add a per user site-package to sys.path + """Add the per-user site-packages directory, if enabled. - Each user has its own python directory with site-packages in the - home directory. + The user site directory is added only when user site-packages are enabled + and the directory exists. Return *known_paths*, updated with any paths + added by addsitedir(). """ - # get the per user site-package path - # this call will also make sure USER_BASE and USER_SITE are set - _trace("Processing user site-packages") - user_site = getusersitepackages() - - if ENABLE_USER_SITE and os.path.isdir(user_site): - addsitedir(user_site, known_paths) + state = StartupState(known_paths) + state.addusersitepackages() + state.process() return known_paths + def getsitepackages(prefixes=None): """Returns a list containing all global site-packages directories. @@ -421,15 +774,20 @@ def getsitepackages(prefixes=None): sitepackages.append(os.path.join(prefix, "Lib", "site-packages")) return sitepackages -def addsitepackages(known_paths, prefixes=None): - """Add site-packages to sys.path""" - _trace("Processing global site-packages") - for sitedir in getsitepackages(prefixes): - if os.path.isdir(sitedir): - addsitedir(sitedir, known_paths) +def addsitepackages(known_paths, prefixes=None): + """Add global site-packages directories, if they exist. + + Site-packages directories are computed from *prefixes*, or from the global + prefixes when *prefixes* is None. Return *known_paths*, updated with any + paths added by addsitedir(). + """ + state = StartupState(known_paths) + state.addsitepackages(prefixes) + state.process() return known_paths + def setquit(): """Define new builtins 'quit' and 'exit'. @@ -532,6 +890,8 @@ def register_readline(): import _pyrepl.unix_console console_errors = _pyrepl.unix_console._error from _pyrepl.main import CAN_USE_PYREPL + except ModuleNotFoundError: + CAN_USE_PYREPL = False finally: sys.path = original_path except ImportError: @@ -594,6 +954,15 @@ def register_readline(): def venv(known_paths): + """Process pyvenv.cfg and add the venv site-packages, if applicable.""" + state = StartupState(known_paths) + _venv(state) + state.process() + return known_paths + + +def _venv(state): + """State-driven implementation of venv(); used by main() for batching.""" global PREFIXES, ENABLE_USER_SITE env = os.environ @@ -633,20 +1002,22 @@ def venv(known_paths): sys._home = value if sys.prefix != site_prefix: - _warn(f'Unexpected value in sys.prefix, expected {site_prefix}, got {sys.prefix}', RuntimeWarning) + _warn( + f'Unexpected value in sys.prefix, expected {site_prefix}, got {sys.prefix}', + RuntimeWarning) if sys.exec_prefix != site_prefix: - _warn(f'Unexpected value in sys.exec_prefix, expected {site_prefix}, got {sys.exec_prefix}', RuntimeWarning) + _warn( + f'Unexpected value in sys.exec_prefix, expected {site_prefix}, got {sys.exec_prefix}', + RuntimeWarning) - # Doing this here ensures venv takes precedence over user-site - addsitepackages(known_paths, [sys.prefix]) + # Doing this here ensures venv takes precedence over user-site. + state.addsitepackages([sys.prefix]) if system_site == "true": PREFIXES += [sys.base_prefix, sys.base_exec_prefix] else: ENABLE_USER_SITE = False - return known_paths - def execsitecustomize(): """Run custom site specific code, if available.""" @@ -697,17 +1068,25 @@ def main(): global ENABLE_USER_SITE orig_path = sys.path[:] - known_paths = removeduppaths() + removeduppaths() if orig_path != sys.path: # removeduppaths() might make sys.path absolute. - # fix __file__ and __cached__ of already imported modules too. + # Fix __file__ of already imported modules too. abs_paths() - known_paths = venv(known_paths) + state = StartupState(set()) + _venv(state) + if ENABLE_USER_SITE is None: ENABLE_USER_SITE = check_enableusersite() - known_paths = addusersitepackages(known_paths) - known_paths = addsitepackages(known_paths) + + state.addusersitepackages() + state.addsitepackages() + # PEP 829: flush accumulated data from all .pth and .start files. + # Paths are extended first, then deprecated import lines are exec'd, + # and finally .start entry points are executed — ensuring sys.path is + # fully populated before any startup code runs. + state.process() setquit() setcopyright() sethelper() diff --git a/Lib/test/test_site.py b/Lib/test/test_site.py index 3c6d48568..548c90260 100644 --- a/Lib/test/test_site.py +++ b/Lib/test/test_site.py @@ -16,6 +16,7 @@ from test.support.os_helper import TESTFN, EnvironmentVarGuard from test.support.script_helper import spawn_python, kill_python import ast import builtins +import contextlib import glob import io import os @@ -27,6 +28,7 @@ import sys import sysconfig import tempfile from textwrap import dedent +from types import SimpleNamespace import urllib.error import urllib.request from unittest import mock @@ -123,14 +125,11 @@ class HelperFunctionsTests(unittest.TestCase): # comment or import that is a valid directory name for where the .pth # file resides; invalid directories are not added pth_file = PthFile() - pth_file.cleanup(prep=True) # to make sure that nothing is - # pre-existing that shouldn't be - try: - pth_file.create() + # Ensure we have a clean slate. + pth_file.cleanup(prep=True) + with pth_file.create(): site.addpackage(pth_file.base_dir, pth_file.filename, set()) self.pth_file_tests(pth_file) - finally: - pth_file.cleanup() def make_pth(self, contents, pth_dir='.', pth_name=TESTFN): # Create a .pth file and return its (abspath, basename). @@ -150,9 +149,6 @@ class HelperFunctionsTests(unittest.TestCase): self.assertRegex(err_out.getvalue(), "line 1") self.assertRegex(err_out.getvalue(), re.escape(os.path.join(pth_dir, pth_fn))) - # XXX: the previous two should be independent checks so that the - # order doesn't matter. The next three could be a single check - # but my regex foo isn't good enough to write it. self.assertRegex(err_out.getvalue(), 'Traceback') self.assertRegex(err_out.getvalue(), r'import bad-syntax') self.assertRegex(err_out.getvalue(), 'SyntaxError') @@ -162,10 +158,8 @@ class HelperFunctionsTests(unittest.TestCase): pth_dir, pth_fn = self.make_pth("randompath\nimport nosuchmodule\n") with captured_stderr() as err_out: site.addpackage(pth_dir, pth_fn, set()) - self.assertRegex(err_out.getvalue(), "line 2") self.assertRegex(err_out.getvalue(), re.escape(os.path.join(pth_dir, pth_fn))) - # XXX: ditto previous XXX comment. self.assertRegex(err_out.getvalue(), 'Traceback') self.assertRegex(err_out.getvalue(), 'ModuleNotFoundError') @@ -178,65 +172,97 @@ class HelperFunctionsTests(unittest.TestCase): def test_addpackage_import_bad_pth_file(self): # Issue 5258 pth_dir, pth_fn = self.make_pth("abc\x00def\n") - with captured_stderr() as err_out: - self.assertFalse(site.addpackage(pth_dir, pth_fn, set())) - self.maxDiff = None - self.assertEqual(err_out.getvalue(), "") for path in sys.path: if isinstance(path, str): self.assertNotIn("abc\x00def", path) def test_addsitedir(self): - # Same tests for test_addpackage since addsitedir() essentially just - # calls addpackage() for every .pth file in the directory + # addsitedir() reads .pth files and, when called standalone + # (known_paths=None), flushes paths and import lines immediately. pth_file = PthFile() - pth_file.cleanup(prep=True) # Make sure that nothing is pre-existing - # that is tested for - try: - pth_file.create() - site.addsitedir(pth_file.base_dir, set()) + # Ensure we have a clean slate. + pth_file.cleanup(prep=True) + with pth_file.create(): + site.addsitedir(pth_file.base_dir) + self.pth_file_tests(pth_file) + + def test_addsitedir_explicit_flush(self): + # StartupState.addsitedir() reads .pth files and accumulates pending + # state without flushing. A subsequent state.process() call then + # applies the paths and runs the import lines. + pth_file = PthFile() + # Ensure we have a clean slate. + pth_file.cleanup(prep=True) + with pth_file.create(): + state = site.StartupState(known_paths=set()) + state.addsitedir(pth_file.base_dir) + self.assertNotIn(pth_file.imported, sys.modules) + state.process() self.pth_file_tests(pth_file) - finally: - pth_file.cleanup() def test_addsitedir_dotfile(self): pth_file = PthFile('.dotfile') + # Ensure we have a clean slate. pth_file.cleanup(prep=True) - try: - pth_file.create() - site.addsitedir(pth_file.base_dir, set()) + with pth_file.create(): + site.addsitedir(pth_file.base_dir) self.assertNotIn(site.makepath(pth_file.good_dir_path)[0], sys.path) self.assertIn(pth_file.base_dir, sys.path) - finally: - pth_file.cleanup() @unittest.skipUnless(hasattr(os, 'chflags'), 'test needs os.chflags()') def test_addsitedir_hidden_flags(self): pth_file = PthFile() + # Ensure we have a clean slate. pth_file.cleanup(prep=True) - try: - pth_file.create() + with pth_file.create(): st = os.stat(pth_file.file_path) os.chflags(pth_file.file_path, st.st_flags | stat.UF_HIDDEN) - site.addsitedir(pth_file.base_dir, set()) + site.addsitedir(pth_file.base_dir) self.assertNotIn(site.makepath(pth_file.good_dir_path)[0], sys.path) self.assertIn(pth_file.base_dir, sys.path) - finally: - pth_file.cleanup() @unittest.skipUnless(sys.platform == 'win32', 'test needs Windows') @support.requires_subprocess() def test_addsitedir_hidden_file_attribute(self): pth_file = PthFile() + # Ensure we have a clean slate. pth_file.cleanup(prep=True) - try: - pth_file.create() + with pth_file.create(): subprocess.check_call(['attrib', '+H', pth_file.file_path]) - site.addsitedir(pth_file.base_dir, set()) + site.addsitedir(pth_file.base_dir) self.assertNotIn(site.makepath(pth_file.good_dir_path)[0], sys.path) self.assertIn(pth_file.base_dir, sys.path) - finally: - pth_file.cleanup() + + def test_sitedir_variable(self): + # gh-149671: setuptools use of `-nspkg.pth` files in Python < 3.15. + code = '; '.join(( + # Code used by "-nspkg.pth" files generated by setuptools. + "import sys", + "sitedir = sys._getframe(1).f_locals['sitedir']", + "print(sitedir)", + )) + pth_dir, pth_fn = self.make_pth(code) + with support.captured_stdout() as stdout: + known_paths = site.addpackage(pth_dir, pth_fn, set()) + sitedir = stdout.getvalue().rstrip() + self.assertEqual(sitedir, pth_dir) + + def test_fullname_variable(self): + # gh-156404: ".pth" files generated before Python 3.15 read the path + # of the file being processed from the frame executing the import + # line, under the name "fullname". For example, "wheel-axle" + # generates: "import wheel_axle.runtime; + # wheel_axle.runtime.finalize(fullname);" + code = '; '.join(( + "import sys", + "print(fullname)", + "print(filename)", + )) + pth_dir, pth_fn = self.make_pth(code) + with support.captured_stdout() as stdout: + site.addpackage(pth_dir, pth_fn, set()) + expected = os.path.join(pth_dir, pth_fn) + self.assertEqual(stdout.getvalue().splitlines(), [expected, expected]) # This tests _getuserbase, hence the double underline # to distinguish from a test for getuserbase @@ -402,7 +428,7 @@ class HelperFunctionsTests(unittest.TestCase): self.assertEqual(sys.stderr.getvalue(), out) -class PthFile(object): +class PthFile: """Helper class for handling testing of .pth files""" def __init__(self, filename_base=TESTFN, imported="time", @@ -417,6 +443,7 @@ class PthFile(object): self.good_dir_path = os.path.join(self.base_dir, self.good_dirname) self.bad_dir_path = os.path.join(self.base_dir, self.bad_dirname) + @contextlib.contextmanager def create(self): """Create a .pth file with a comment, blank lines, an ``import ``, a line with self.good_dirname, and a line with @@ -425,19 +452,21 @@ class PthFile(object): Creation of the directory for self.good_dir_path (based off of self.good_dirname) is also performed. - Make sure to call self.cleanup() to undo anything done by this method. - + Used as a context manager: self.cleanup() is called on exit. """ - FILE = open(self.file_path, 'w') - try: - print("#import @bad module name", file=FILE) - print("\n", file=FILE) - print("import %s" % self.imported, file=FILE) - print(self.good_dirname, file=FILE) - print(self.bad_dirname, file=FILE) - finally: - FILE.close() + with open(self.file_path, 'w') as fp: + print(f"""\ +#import @bad module name +import {self.imported} +{self.good_dirname} +{self.bad_dirname} +""", file=fp) + os.mkdir(self.good_dir_path) + try: + yield self + finally: + self.cleanup() def cleanup(self, prep=False): """Make sure that the .pth file is deleted, self.imported is not in @@ -457,6 +486,7 @@ class PthFile(object): if os.path.exists(self.bad_dir_path): os.rmdir(self.bad_dir_path) + class ImportSideEffectTests(unittest.TestCase): """Test side-effects from importing 'site'.""" @@ -468,17 +498,6 @@ class ImportSideEffectTests(unittest.TestCase): """Restore sys.path""" sys.path[:] = self.sys_path - def test_abs_paths_cached_None(self): - """Test for __cached__ is None. - - Regarding to PEP 3147, __cached__ can be None. - - See also: https://bugs.python.org/issue30167 - """ - sys.modules['test'].__cached__ = None - site.abs_paths() - self.assertIsNone(sys.modules['test'].__cached__) - def test_no_duplicate_paths(self): # No duplicate paths should exist in sys.path # Handled by removeduppaths() @@ -557,7 +576,6 @@ class ImportSideEffectTests(unittest.TestCase): output = subprocess.check_output([sys.executable, '-s', '-c', '""']) self.assertNotIn(eyecatcher, output.decode('utf-8')) - @unittest.skipUnless(hasattr(urllib.request, "HTTPSHandler"), 'need SSL support to download license') @test.support.requires_resource('network') @@ -810,6 +828,15 @@ class _pthFileTests(unittest.TestCase): )], env=env) self.assertTrue(rc, "sys.path is incorrect") + @support.requires_subprocess() + def test_underpth_no_user_site(self): + pth_lines = [test.support.STDLIB_DIR, 'import site'] + exe_file = self._create_underpth_exe(pth_lines) + p = subprocess.run([exe_file, '-X', 'utf8', '-c', + 'import sys; ' + 'sys.exit(not sys.flags.no_user_site)']) + self.assertEqual(p.returncode, 0, "sys.flags.no_user_site was 0") + class CommandLineTests(unittest.TestCase): def exists(self, path): @@ -916,5 +943,837 @@ class CommandLineTests(unittest.TestCase): self.assertEqual(output, excepted_output) +class StartFileTests(unittest.TestCase): + """Tests for .start file processing (PEP 829).""" + + def setUp(self): + self.enterContext(import_helper.DirsOnSysPath()) + self.tmpdir = self.sitedir = self.enterContext(os_helper.temp_dir()) + # Each test gets its own StartupState to batch the parsing and + # explicitly invoke the processing. Seed with an empty known_paths + # so dedup is not influenced by the current sys.path. + self.state = site.StartupState(known_paths=set()) + + def _make_start(self, content, name='testpkg', basedir=None): + """Write a .start file and return its basename. + + ``basedir`` defaults to ``self.tmpdir``. Pass an explicit directory + when the .start file needs to live somewhere other than the test's + primary tmpdir (e.g. a nested user-site). + """ + basename = f"{name}.start" + filepath = os.path.join(self.tmpdir if basedir is None else basedir, basename) + with open(filepath, 'w', encoding='utf-8') as f: + f.write(content) + return basename + + def _make_pth(self, content, name='testpkg', basedir=None): + """Write a .pth file and return its basename. + + ``basedir`` defaults to ``self.tmpdir``. Pass an explicit directory + when the .pth file needs to live somewhere other than the test's + primary tmpdir (e.g. a nested user-site). + """ + basename = f"{name}.pth" + filepath = os.path.join(self.tmpdir if basedir is None else basedir, basename) + with open(filepath, 'w', encoding='utf-8') as f: + f.write(content) + return basename + + def _make_mod(self, contents, name='mod', *, package=False, on_path=False): + """Write an importable module (or package), returning its parent dir.""" + extdir = os.path.join(self.sitedir, 'extdir') + os.makedirs(extdir, exist_ok=True) + + # Put the code in a package's dunder-init or flat module. + if package: + pkgdir = os.path.join(extdir, name) + os.mkdir(pkgdir) + modpath = os.path.join(pkgdir, '__init__.py') + else: + modpath = os.path.join(extdir, f'{name}.py') + + with open(modpath, 'w') as fp: + fp.write(contents) + + self.addCleanup(sys.modules.pop, name, None) + if on_path: + # Don't worry, DirsOnSysPath() in setUp() will clean this up. + sys.path.insert(0, extdir) + return extdir + + def _all_entrypoints(self, state=None): + """Flatten state._entrypoints into a list of (filename, entry) tuples.""" + result = [] + state = self.state if state is None else state + for filename, entries in state._entrypoints.items(): + for entry in entries: + result.append((filename, entry)) + return result + + def _just_entrypoints(self, state=None): + return [entry for filename, entry in self._all_entrypoints(state)] + + # There are two classes of tests here. Tests that start with `test_impl_` + # know details about the implementation and they access non-public methods + # and data structures to perform focused functional tests. + # + # Tests that start with `test_addsitedir_` are end-to-end tests that ensure + # integration semantics and functionality as a caller of the public + # surfaces would see. + + # --- Basic StartupState implementation tests --- + + def test_impl_startupstate_defaults_to_sys_path(self): + sys.path.insert(0, self.sitedir) + state = site.StartupState() + self.assertIn(site.makepath(self.sitedir)[1], state._known_paths) + + def test_impl_startupstate_uses_supplied_known_paths(self): + known_paths = set() + state = site.StartupState(known_paths) + state.addsitedir(self.sitedir) + self.assertIs(state._known_paths, known_paths) + self.assertIn(site.makepath(self.sitedir)[1], known_paths) + + # --- StartupState._read_start_file tests --- + + def test_impl_read_start_file_basic(self): + self._make_start("os.path:join\n", name='foo') + self.state._read_start_file(self.sitedir, 'foo.start') + fullname = os.path.join(self.sitedir, 'foo.start') + self.assertEqual( + self.state._entrypoints[fullname], ['os.path:join'] + ) + + def test_impl_read_start_file_multiple_entries(self): + self._make_start("os.path:join\nos.path:exists\n", name='foo') + self.state._read_start_file(self.sitedir, 'foo.start') + fullname = os.path.join(self.sitedir, 'foo.start') + self.assertEqual( + self.state._entrypoints[fullname], + ['os.path:join', 'os.path:exists'], + ) + + def test_impl_read_start_file_comments_and_blanks(self): + self._make_start("# a comment\n\nos.path:join\n \n", name='foo') + self.state._read_start_file(self.sitedir, 'foo.start') + fullname = os.path.join(self.sitedir, 'foo.start') + self.assertEqual( + self.state._entrypoints[fullname], ['os.path:join'] + ) + + def test_impl_read_start_file_accepts_all_non_blank_lines(self): + # Syntax validation is deferred to entry-point execution time + # (where pkgutil.resolve_name(strict=True) enforces the strict + # pkg.mod:callable form), so parsing accepts every non-blank, + # non-comment line, including syntactically invalid ones. + content = ( + "os.path\n" # no colon + "pkg.mod:\n" # empty callable + ":callable\n" # empty module + "pkg.mod:callable:extra\n" # multiple colons + "os.path:join\n" # valid + ) + self._make_start(content, name='foo') + self.state._read_start_file(self.sitedir, 'foo.start') + fullname = os.path.join(self.sitedir, 'foo.start') + self.assertEqual(self.state._entrypoints[fullname], [ + 'os.path', + 'pkg.mod:', + ':callable', + 'pkg.mod:callable:extra', + 'os.path:join', + ]) + + def test_impl_read_start_file_empty(self): + # PEP 829: an empty .start file is still registered as present + # (with an empty entry point list) so that it suppresses `import` + # lines in any matching .pth file. + self._make_start("", name='foo') + self.state._read_start_file(self.sitedir, 'foo.start') + fullname = os.path.join(self.sitedir, 'foo.start') + self.assertEqual(self.state._entrypoints, {fullname: []}) + + def test_impl_read_start_file_comments_only(self): + # As with an empty file, a comments-only .start file is registered + # as present so it can suppress matching .pth `import` lines. + self._make_start("# just a comment\n# another\n", name='foo') + self.state._read_start_file(self.sitedir, 'foo.start') + fullname = os.path.join(self.sitedir, 'foo.start') + self.assertEqual(self.state._entrypoints, {fullname: []}) + + def test_impl_read_start_file_nonexistent(self): + with captured_stderr(): + self.state._read_start_file(self.tmpdir, 'nonexistent.start') + self.assertEqual(self.state._entrypoints, {}) + + @unittest.skipUnless(hasattr(os, 'chflags'), 'test needs os.chflags()') + def test_impl_read_start_file_hidden_flags(self): + self._make_start("os.path:join\n", name='foo') + filepath = os.path.join(self.tmpdir, 'foo.start') + st = os.stat(filepath) + os.chflags(filepath, st.st_flags | stat.UF_HIDDEN) + self.state._read_start_file(self.sitedir, 'foo.start') + self.assertEqual(self.state._entrypoints, {}) + + def test_impl_one_start_file_with_duplicates_not_deduplicated(self): + # PEP 829: duplicate entry points are NOT deduplicated. + self._make_start("os.path:join\nos.path:join\n", name='foo') + self.state._read_start_file(self.sitedir, 'foo.start') + fullname = os.path.join(self.sitedir, 'foo.start') + self.assertEqual( + self.state._entrypoints[fullname], + ['os.path:join', 'os.path:join'], + ) + + def test_impl_two_start_files_with_duplicates_not_deduplicated(self): + self._make_start("os.path:join", name="foo") + self._make_start("os.path:join", name="bar") + self.state._read_start_file(self.sitedir, 'foo.start') + self.state._read_start_file(self.sitedir, 'bar.start') + self.assertEqual( + self._just_entrypoints(), + ['os.path:join', 'os.path:join'], + ) + + def test_impl_read_start_file_accepts_utf8_bom(self): + # PEP 829: .start files MUST be utf-8-sig (UTF-8 with optional BOM). + filepath = os.path.join(self.tmpdir, 'foo.start') + with open(filepath, 'wb') as f: + f.write(b'\xef\xbb\xbf' + b'os.path:join\n') + self.state._read_start_file(self.sitedir, 'foo.start') + fullname = os.path.join(self.sitedir, 'foo.start') + self.assertEqual( + self.state._entrypoints[fullname], ['os.path:join'] + ) + + def test_impl_read_start_file_invalid_utf8_silently_skipped(self): + # PEP 829: .start files MUST be utf-8-sig. Unlike .pth files, there + # is no locale-encoding fallback. A .start file that is not valid + # UTF-8 is silently skipped, with no key registered in + # state._entrypoints and no output to stderr (parsing errors are + # reported only under -v). + filepath = os.path.join(self.tmpdir, 'foo.start') + with open(filepath, 'wb') as f: + # Bare continuation byte -- invalid as a UTF-8 start byte. + f.write(b'\x80\x80\x80\n') + with captured_stderr() as err: + self.state._read_start_file(self.sitedir, 'foo.start') + self.assertEqual(self.state._entrypoints, {}) + self.assertEqual(err.getvalue(), "") + + # --- StartupState._read_pth_file tests --- + + def test_impl_read_pth_file_paths(self): + subdir = os.path.join(self.sitedir, 'mylib') + os.mkdir(subdir) + self._make_pth("mylib\n", name='foo') + self.state._read_pth_file(self.sitedir, 'foo.pth') + fullname = os.path.join(self.sitedir, 'foo.pth') + self.assertIn((fullname, subdir), self.state._path_entries) + + def test_impl_read_pth_file_imports_collected(self): + self._make_pth("import sys\n", name='foo') + self.state._read_pth_file(self.sitedir, 'foo.pth') + fullname = os.path.join(self.sitedir, 'foo.pth') + self.assertEqual( + self.state._importexecs[fullname], ['import sys'] + ) + + def test_impl_read_pth_file_comments_and_blanks(self): + self._make_pth("# comment\n\n \n", name='foo') + self.state._read_pth_file(self.sitedir, 'foo.pth') + self.assertEqual(self.state._path_entries, []) + self.assertEqual(self.state._importexecs, {}) + + def test_impl_read_pth_file_deduplication(self): + subdir = os.path.join(self.sitedir, 'mylib') + os.mkdir(subdir) + # self.state._known_paths acts as the deduplication ledger across + # both reads. + self._make_pth("mylib\n", name='a') + self._make_pth("mylib\n", name='b') + self.state._read_pth_file(self.sitedir, 'a.pth') + self.state._read_pth_file(self.sitedir, 'b.pth') + # There is only one entry across both files. + all_dirs = [dir_ for filename, dir_ in self.state._path_entries] + self.assertEqual(all_dirs, [subdir]) + + def test_impl_read_pth_file_bad_line_continues(self): + # PEP 829: errors on individual lines don't abort processing the file. + subdir = os.path.join(self.sitedir, 'goodpath') + os.mkdir(subdir) + self._make_pth("abc\x00def\ngoodpath\n", name='foo') + with captured_stderr(): + self.state._read_pth_file(self.sitedir, 'foo.pth') + fullname = os.path.join(self.sitedir, 'foo.pth') + self.assertIn((fullname, subdir), self.state._path_entries) + + def _flags_with_verbose(self, verbose): + # Build a sys.flags clone with verbose overridden but every + # other field preserved, so unrelated reads like + # sys.flags.optimize during io.open_code() continue to work. + attrs = { + name: getattr(sys.flags, name) + for name in sys.flags.__match_args__ + } + attrs['verbose'] = verbose + return SimpleNamespace(**attrs) + + def test_impl_read_pth_file_parse_error_silent_by_default(self): + # PEP 829: parse-time errors are silent unless -v is given. + # Force the error path by making makepath() raise an exception. + self._make_pth("badline\n", name='foo') + with ( + mock.patch('site.makepath', side_effect=ValueError("boom")), + mock.patch('sys.flags', self._flags_with_verbose(False)), + captured_stderr() as err, + ): + self.state._read_pth_file(self.sitedir, 'foo.pth') + self.assertEqual(err.getvalue(), "") + + def test_impl_read_pth_file_parse_error_reported_under_verbose(self): + # PEP 829: parse-time errors are reported when -v is given. + self._make_pth("badline\n", name='foo') + with ( + mock.patch('site.makepath', side_effect=ValueError("boom")), + mock.patch('sys.flags', self._flags_with_verbose(True)), + captured_stderr() as err, + ): + self.state._read_pth_file(self.sitedir, 'foo.pth') + out = err.getvalue() + self.assertIn('Error in', out) + self.assertIn('foo.pth', out) + + def test_impl_read_pth_file_locale_fallback(self): + # PEP 829: .pth files that fail UTF-8 decoding fall back to the + # locale encoding for backward compatibility (deprecated in + # 3.15, to be removed in 3.20). Mock locale.getencoding() so + # the test does not depend on the host's actual locale. + subdir = os.path.join(self.sitedir, 'mylib') + os.mkdir(subdir) + filepath = os.path.join(self.tmpdir, 'foo.pth') + # \xe9 is invalid UTF-8 but valid in latin-1. + with open(filepath, 'wb') as f: + f.write(b'# caf\xe9 comment\nmylib\n') + with ( + mock.patch('locale.getencoding', return_value='latin-1'), + captured_stderr(), + ): + self.state._read_pth_file(self.sitedir, 'foo.pth') + fullname = os.path.join(self.sitedir, 'foo.pth') + self.assertIn((fullname, subdir), self.state._path_entries) + + # --- StartupState._execute_start_entrypoints tests --- + + def test_impl_execute_entrypoints_with_callable(self): + # An entry point with a callable. + self._make_mod("""\ +called = False +def startup(): + global called + called = True +""", name='epmod', package=True, on_path=True) + fullname = os.path.join(self.sitedir, 'epmod.start') + self.state._entrypoints[fullname] = ['epmod:startup'] + self.state._execute_start_entrypoints() + import epmod + self.assertTrue(epmod.called) + + def test_impl_execute_entrypoints_import_error(self): + # Import errors print a traceback and continue. + fullname = os.path.join(self.sitedir, 'bad.start') + self.state._entrypoints[fullname] = [ + 'nosuchmodule_xyz:func', 'os.path:join', + ] + with captured_stderr() as err: + self.state._execute_start_entrypoints() + self.assertIn('nosuchmodule_xyz', err.getvalue()) + # os.path:join should still have been called (no exception for it) + + def test_impl_execute_entrypoints_strict_syntax_rejection(self): + # PEP 829: only the strict pkg.mod:callable form is valid. At entry + # point execution time, pkgutil.resolve_name(strict=True) raises a + # ValueError for the invalid syntax. The invalid entry is reported + # and execution continues with the next one. + fullname = os.path.join(self.sitedir, 'bad.start') + self.state._entrypoints[fullname] = [ + 'os.path', # no colon + 'pkg.mod:', # empty callable + ':callable', # empty module + 'pkg.mod:callable:extra', # multiple colons + ] + with captured_stderr() as err: + self.state._execute_start_entrypoints() + out = err.getvalue() + self.assertIn('Invalid entry point syntax', out) + for bad in ( + 'os.path', + 'pkg.mod:', + ':callable', + 'pkg.mod:callable:extra', + ): + self.assertIn(bad, out) + + def test_impl_execute_entrypoints_callable_error(self): + # A callable that errors prints a traceback but continues. + self._make_mod("""\ +def fail(): + raise RuntimeError("boom") +""", name='badmod', package=True, on_path=True) + fullname = os.path.join(self.sitedir, 'badmod.start') + self.state._entrypoints[fullname] = ['badmod:fail'] + with captured_stderr() as err: + self.state._execute_start_entrypoints() + self.assertIn('RuntimeError', err.getvalue()) + self.assertIn('boom', err.getvalue()) + + def test_impl_execute_entrypoints_duplicates_called_twice(self): + # PEP 829: duplicate entry points execute multiple times. + self._make_mod("""\ +call_count = 0 +def bump(): + global call_count + call_count += 1 +""", name='countmod', package=False, on_path=True) + fullname = os.path.join(self.sitedir, 'countmod.start') + self.state._entrypoints[fullname] = [ + 'countmod:bump', 'countmod:bump', + ] + self.state._execute_start_entrypoints() + import countmod + self.assertEqual(countmod.call_count, 2) + + # --- StartupState._exec_imports tests --- + + def test_impl_exec_imports_suppressed_by_matching_start(self): + # Import lines from foo.pth are suppressed when foo.start exists. + self._make_mod("""\ +call_count = 0 +def bump(incr=2): + global call_count + call_count += incr +""", name='countmod', package=False, on_path=True) + self._make_start("countmod:bump\n", name='foo') + self._make_pth("import countmod; countmod.bump(1)\n", name='foo') + self.state._read_pth_file(self.sitedir, 'foo.pth') + self.state._read_start_file(self.sitedir, 'foo.start') + self.state._exec_imports() + self.state._execute_start_entrypoints() + import countmod + # This will be 2 because the entry point is called with no + # arguments, and the .pth import line is never exec'd. + self.assertEqual(countmod.call_count, 2) + + def test_impl_exec_imports_not_suppressed_by_different_start(self): + # Import lines from foo.pth are NOT suppressed by bar.start. + self._make_mod("""\ +call_count = 0 +def bump(): + global call_count + call_count += 1 +""", name='countmod', package=False, on_path=True) + pth_fullname = os.path.join(self.sitedir, 'foo.pth') + start_fullname = os.path.join(self.sitedir, 'bar.start') + self.state._importexecs[pth_fullname] = ['import countmod; countmod.bump()'] + self.state._entrypoints[start_fullname] = ['os.path:join'] + self.state._exec_imports() + import countmod + self.assertEqual(countmod.call_count, 1) + + def test_impl_exec_imports_suppressed_by_empty_matching_start(self): + self._make_start("", name='foo') + self._make_pth("import epmod; epmod.startup()", name='foo') + self._make_mod("""\ +called = False +def startup(): + global called + called = True +""", name='epmod', package=True, on_path=True) + self.state._read_pth_file(self.sitedir, 'foo.pth') + self.state._read_start_file(self.sitedir, 'foo.start') + self.state._exec_imports() + import epmod + self.assertFalse(epmod.called) + + # --- StartupState._extend_syspath tests --- + + def test_impl_extend_syspath_existing_dir(self): + subdir = os.path.join(self.sitedir, 'extlib') + os.mkdir(subdir) + self.state._path_entries.append(('test.pth', subdir)) + self.state._extend_syspath() + self.assertIn(subdir, sys.path) + + def test_impl_extend_syspath_nonexistent_dir(self): + nonesuch = os.path.join(self.sitedir, 'nosuchdir') + self.state._path_entries.append(('test.pth', nonesuch)) + with captured_stderr() as err: + self.state._extend_syspath() + self.assertNotIn(nonesuch, sys.path) + self.assertIn('does not exist', err.getvalue()) + + # --- addsitedir integration tests --- + + def test_addsitedir_pth_import_skipped_when_matching_start_exists(self): + # PEP 829: an empty .start file disables the matching .pth's import + # lines, even when the .start has no entry points of its own. + self._make_mod("flag = False\n", name='suppressed', on_path=True) + self._make_start("", name='foo') + self._make_pth( + "import suppressed; suppressed.flag = True\n", + name='foo') + site.addsitedir(self.sitedir, set()) + import suppressed + self.assertFalse( + suppressed.flag, + "import line in foo.pth should be suppressed by foo.start") + + def test_addsitedir_dotfile_start_entrypoint_not_executed(self): + # .start files starting with '.' are skipped, so their entry + # points must not run. + self._make_mod("""\ +called = False +def hook(): + global called + called = True +""", + name='dotted', on_path=True) + self._make_start("dotted:hook\n", name='.hidden') + site.addsitedir(self.sitedir, set()) + import dotted + self.assertFalse(dotted.called) + + def test_addsitedir_dedups_paths_across_pth_files(self): + # PEP 829: when multiple .pth files reference the same path within + # a single addsitedir() invocation, the path is appended to + # sys.path exactly once. + subdir = os.path.join(self.sitedir, 'shared') + os.mkdir(subdir) + self._make_pth("shared\n", name='a') + self._make_pth("shared\n", name='b') + before = sys.path.count(subdir) + site.addsitedir(self.sitedir, set()) + self.assertEqual(sys.path.count(subdir), before + 1) + + def test_addsitedir_discovers_start_files(self): + # addsitedir() should discover .start files and accumulate entries. + self._make_start("os.path:join\n", name='foo') + state = site.StartupState(known_paths=set()) + state.addsitedir(self.sitedir) + fullname = os.path.join(self.sitedir, 'foo.start') + self.assertIn( + 'os.path:join', state._entrypoints[fullname] + ) + + def test_addsitedir_pth_paths_still_work_with_start(self): + # Path lines in .pth files still work even when a .start file exists. + subdir = os.path.join(self.sitedir, 'mylib') + os.mkdir(subdir) + self._make_start("os.path:join\n", name='foo') + self._make_pth("mylib\n", name='foo') + state = site.StartupState(known_paths=set()) + state.addsitedir(self.sitedir) + fullname = os.path.join(self.sitedir, 'foo.pth') + self.assertIn((fullname, subdir), state._path_entries) + + def test_addsitedir_start_alphabetical_order(self): + # Multiple .start files are discovered alphabetically. + self._make_start("os.path:join\n", name='zzz') + self._make_start("os.path:exists\n", name='aaa') + state = site.StartupState(known_paths=set()) + state.addsitedir(self.sitedir) + entries = self._just_entrypoints(state) + idx_a = entries.index('os.path:exists') + idx_z = entries.index('os.path:join') + self.assertLess(idx_a, idx_z) + + def test_addsitedir_pth_and_start(self): + # Create a .pth and .start with the same basename; verify both the + # .pth data and .start data is collected. + subdir = os.path.join(self.sitedir, 'mylib') + os.mkdir(subdir) + self._make_pth("mylib\n", name='foo') + self._make_start("os.path:join\n", name='foo') + state = site.StartupState(known_paths=set()) + state.addsitedir(self.sitedir) + # Both should be collected. + pth_fullname = os.path.join(self.sitedir, 'foo.pth') + start_fullname = os.path.join(self.sitedir, 'foo.start') + self.assertIn((pth_fullname, subdir), state._path_entries) + self.assertIn( + 'os.path:join', + state._entrypoints.get(start_fullname, []), + ) + + def test_impl_addsitedir_skips_dotfile_start(self): + # .start files starting with '.' are skipped. + # This will create `.hidden.start`. + self._make_start("os.path:join\n", name='.hidden') + state = site.StartupState(known_paths=set()) + state.addsitedir(self.sitedir) + self.assertEqual(state._entrypoints, {}) + + def test_addsitedir_standalone_flushes(self): + # Standalone addsitedir creates a per-call StartupState and processes + # it before returning, so the caller sees the effect immediately. + subdir = os.path.join(self.sitedir, 'flushlib') + os.mkdir(subdir) + self._make_pth("flushlib\n", name='foo') + # No arguments means state is implied and processing is eager. + site.addsitedir(self.sitedir) + self.assertIn(subdir, sys.path) + + def test_addsitedir_explicit_startup_state_does_not_flush(self): + # With an explicit StartupState, addsitedir accumulates pending state + # but does not flush it; sys.path is updated only when process() is + # called explicitly. + subdir = os.path.join(self.sitedir, 'acclib') + os.mkdir(subdir) + self._make_pth("acclib\n", name='foo') + state = site.StartupState(known_paths=set()) + state.addsitedir(self.sitedir) + # Path is pending, not yet on sys.path. + self.assertNotIn(subdir, sys.path) + fullname = os.path.join(self.sitedir, 'foo.pth') + self.assertIn((fullname, subdir), state._path_entries) + + def test_addsitedir_startup_state_preserves_site_relative_order(self): + # As pointed out by @ncoghlan in + # https://github.com/python/cpython/issues/150228#issuecomment-4528614952 + # a subtle ordering change was inadvertently introduced where the + # interspersing of the sitedirs with the sys.path extensions they defined + # was lost during batch mode. You'd see all the sitedirs, then all path + # extensions. This test ensures that the old interspersing behavior + # has been restored. + # + # Let's start by creating two sitedirs, each with an extension directory + # which will be added to sys.path by .pth files in the respective sitedirs. + sitedir2 = self.enterContext(os_helper.temp_dir()) + extdir1 = os.path.join(self.sitedir, 'ext1') + extdir2 = os.path.join(sitedir2, 'ext2') + os.mkdir(extdir1) + os.mkdir(extdir2) + self._make_pth(extdir1 + "\n", name='one') + self._make_pth(extdir2 + "\n", name='two', basedir=sitedir2) + # Now create an explicit batch, add each sitedir, then process the + # entire batch. + state = site.StartupState(known_paths=set()) + state.addsitedir(self.sitedir) + state.addsitedir(sitedir2) + state.process() + # Ensure that on sys.path we see this interspersed order: + # [sitedir1, extdir1, sitedir2, extdir2] + indexes = [ + sys.path.index(path) + for path in (self.sitedir, extdir1, sitedir2, extdir2) + ] + # If the index ordering is the same, we preserved the intersperse. + self.assertEqual(indexes, sorted(indexes)) + + def test_addsitedir_startup_state_paths_before_entrypoints(self): + # Ensure that sys.path extensions are available by the time + # .start file entry points are called. + extdir = self._make_mod("""\ +called = False +def hook(): + global called + called = True +""") + self.assertNotIn(extdir, sys.path) + self._make_pth("extdir\n", name='extlib') + self._make_start("mod:hook\n", name='extlib') + # Before the startup state is explicitly processed, neither + # the path extension is added, nor the entry point called. + state = site.StartupState(known_paths=set()) + state.addsitedir(self.sitedir) + self.assertNotIn(extdir, sys.path) + self.assertNotIn('mod', sys.modules) + # After processing the batch, sys.path is extended and + # the entry point was called. + state.process() + self.assertIn(extdir, sys.path) + import mod + self.assertTrue(mod.called) + + def test_pth_path_is_available_to_start_entrypoint(self): + # Core PEP 829 invariant: all .pth path extensions are applied to + # sys.path *before* any .start entry point runs, so an entry + # point may live in a module reachable only via a .pth-extended + # path. If the flush phases were inverted, resolving the entry + # point would fail with ModuleNotFoundError. + extdir = self._make_mod("""\ +called = False +def hook(): + global called + called = True +""") + # extdir is not on sys.path; only the .pth file makes it so. + self.assertNotIn(extdir, sys.path) + self._make_pth("extdir\n", name='extlib') + self._make_start("mod:hook\n", name='extlib') + + # Standalone addsitedir() triggers the full flush sequence. + site.addsitedir(self.sitedir) + + self.assertIn(extdir, sys.path) + import mod + self.assertTrue( + mod.called, + "entry point did not run; .pth path was likely not applied " + "before .start entry-point execution") + + # --- bugs --- + + # gh-75723 + def test_addsitdir_idempotent_pth(self): + # Adding the same sitedir twice with a known_paths, should not + # process .pth files twice. + extdir = self._make_mod("""\ +_pth_count = 0 +""") + self._make_pth(f"""\ +{extdir} +import mod; mod._pth_count += 1 +""") + dirs = set() + dirs = site.addsitedir(self.sitedir, dirs) + dirs = site.addsitedir(self.sitedir, dirs) + import mod + self.assertEqual(mod._pth_count, 1) + + def test_addsitdir_idempotent_start(self): + # Adding the same sitedir twice with a known_paths, should not + # process .pth files twice. + extdir = self._make_mod("""\ +_pth_count = 0 +def increment(): + global _pth_count + _pth_count += 1 +""") + self._make_pth(f"""\ +{extdir} +""") + self._make_start("""\ +mod:increment +""") + dirs = set() + dirs = site.addsitedir(self.sitedir, dirs) + dirs = site.addsitedir(self.sitedir, dirs) + import mod + self.assertEqual(mod._pth_count, 1) + + # gh-149504 + def test_reentrant_addsitedir_pth(self): + # An import line in a .pth file that calls site.addsitedir() + # must not crash or re-execute outer entries while the outer + # call is still processing its pending startup state. + overlay = self.enterContext(os_helper.temp_dir()) + overlay_pth = os.path.join(overlay, 'overlay.pth') + pkgdir = self.enterContext(os_helper.temp_dir()) + with open(overlay_pth, 'w', encoding='utf-8') as fp: + print(pkgdir, file=fp) + self._make_pth(f"import site; site.addsitedir({overlay!r})\n") + site.addsitedir(self.sitedir, set()) + self.assertIn(overlay, sys.path) + self.assertIn(pkgdir, sys.path) + + # gh-149504 + def test_reentrant_addsitedir_start(self): + # As above, but the re-entry happens from a .start entry point + # instead of a .pth import line. The entry point execution + # phase is vulnerable to the same class of bug. + overlay = self.enterContext(os_helper.temp_dir()) + overlay_pth = os.path.join(overlay, 'overlay.pth') + pkgdir = self.enterContext(os_helper.temp_dir()) + with open(overlay_pth, 'w', encoding='utf-8') as fp: + print(pkgdir, file=fp) + self._make_mod(f"""\ +import site +def bootstrap(): + site.addsitedir({overlay!r}) +""", + name='reenter_helper', on_path=True) + self._make_start("reenter_helper:bootstrap\n") + site.addsitedir(self.sitedir, set()) + self.assertIn(overlay, sys.path) + self.assertIn(pkgdir, sys.path) + + # gh-149819 + @unittest.skipUnless(site.ENABLE_USER_SITE, "requires user-site") + @support.requires_subprocess() + def test_pth_processed_when_sitedir_already_on_path(self): + # A .pth file in a site-packages directory must still be processed by + # site.main() when that directory is already on sys.path at + # interpreter start up, for example in a subprocess that inherits + # PYTHONPATH from its parent. Before the fix, main() seeded + # known_paths with all entries derived from removeduppaths(), and + # addsitedir() then skipped .pth processing for any directory already + # in known_paths. + user_base = self.tmpdir + user_site = site._get_path(user_base) + os.makedirs(user_site) + sentinel = "GH149819_PTH_RAN" + # Writing some text to stderr is the simplest observable side effect. + self._make_pth(f"""\ +import sys; sys.stderr.write({sentinel!r}); sys.stderr.flush() +""", + name='gh149819', + basedir=user_site) + with EnvironmentVarGuard() as env: + # PYTHONUSERBASE points USER_SITE at our temp directory so + # site.main() will call addsitedir() on it, rather than on the + # host interpreter's real user-site. + env['PYTHONUSERBASE'] = user_base + # PYTHONPATH puts that same directory on sys.path before + # site.main() runs in the subprocess. This is what triggers the + # bug: removeduppaths() records it in known_paths, and the unfixed + # addsitedir() then skips .pth processing. + env['PYTHONPATH'] = user_site + result = subprocess.run( + [sys.executable, '-c', ''], + capture_output=True, + check=True, + ) + self.assertIn(sentinel.encode(), result.stderr) + + @unittest.skipUnless(site.ENABLE_USER_SITE, "requires user-site") + @support.requires_subprocess() + def test_start_processed_when_sitedir_already_on_path(self): + # Companion to test_pth_processed_when_sitedir_already_on_path: + # the same dedup-guard skip in addsitedir() suppressed both .pth + # and .start file processing, so verify .start entry points also + # run for a site-packages directory inherited via PYTHONPATH. + user_base = self.tmpdir + user_site = site._get_path(user_base) + os.makedirs(user_site) + sentinel = "GH149819_START_RAN" + # The .start entry point resolves to a callable, so we write a + # tiny importable module that outputs the sentinel text. It lands in + # /extdir. That path is added to PYTHONPATH below so + # the subprocess can import it. + extdir = self._make_mod(f"""\ +import sys +def run(): + sys.stderr.write({sentinel!r}) + sys.stderr.flush() +""", name='gh149819mod') + self._make_start( + 'gh149819mod:run\n', name='gh149819', basedir=user_site + ) + with EnvironmentVarGuard() as env: + # See above for details. + env['PYTHONUSERBASE'] = user_base + env['PYTHONPATH'] = os.pathsep.join([user_site, extdir]) + result = subprocess.run( + [sys.executable, '-c', ''], + capture_output=True, + check=True, + ) + self.assertIn(sentinel.encode(), result.stderr) + + + if __name__ == "__main__": unittest.main()