19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

30.77

/src/python/pants/init/logging.py

# Copyright 2018 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import http.client
import locale
import logging
import sys
from collections.abc import Iterator
from contextlib import contextmanager
from io import BufferedReader, TextIOWrapper
from logging import Formatter, Handler, LogRecord
from pathlib import PurePath

import pants.util.logging as pants_logging
from pants.engine.internals import native_engine
from pants.option.option_value_container import OptionValueContainer
from pants.util.dirutil import safe_mkdir_for
from pants.util.docutil import doc_url
from pants.util.logging import LogLevel
from pants.util.strutil import strip_prefix

# Although logging supports the WARN level, its not documented and could conceivably be yanked.
# Since pants has supported 'warn' since inception, leave the 'warn' choice as-is but explicitly
# setup a 'WARN' logging level name that maps to 'WARNING'.
logging.addLevelName(logging.WARNING, "WARN")
logging.addLevelName(pants_logging.TRACE, "TRACE")


class _NativeHandler(Handler):
    """This class is installed as a Python logging module handler (using the logging.addHandler
    method) and proxies logs to the Rust logging infrastructure."""

    def emit(self, record: LogRecord) -> None:
        native_engine.write_log(self.format(record), record.levelno, record.name)

    def flush(self) -> None:
        native_engine.flush_log()


class _ExceptionFormatter(Formatter):
    """Possibly render the stacktrace and possibly give debug hints, based on global options."""

    def __init__(self, level: LogLevel, *, print_stacktrace: bool) -> None:
        super().__init__(None)
        self.level = level
        self.print_stacktrace = print_stacktrace

    def formatException(self, exc_info):
        stacktrace = super().formatException(exc_info) if self.print_stacktrace else ""

        debug_instructions = []
        if not self.print_stacktrace:
            debug_instructions.append("--print-stacktrace for more error details")
        if self.level not in {LogLevel.DEBUG, LogLevel.TRACE}:
            debug_instructions.append("-ldebug for more logs")
        debug_instructions = (
            f"Use {' and/or '.join(debug_instructions)}. " if debug_instructions else ""
        )

        return (
            f"{stacktrace}\n\n{debug_instructions}\nSee {doc_url('docs/using-pants/troubleshooting-common-issues')} for common "
            f"issues.\nConsider reaching out for help: {doc_url('community/getting-help')}\n"
        )


@contextmanager
def stdio_destination(stdin_fileno: int, stdout_fileno: int, stderr_fileno: int) -> Iterator[None]:
    """Sets a destination for both logging and stdio: must be called after `initialize_stdio`.

    After `initialize_stdio` and outside of this contextmanager, the default stdio destination is
    the pants.log. But inside of this block, all engine "tasks"/@rules that are spawned will have
    thread/task-local state that directs their IO to the given destination. When the contextmanager
    exits all tasks will be restored to the default destination (regardless of whether they have
    completed).
    """
    if not logging.getLogger(None).handlers:
        raise AssertionError("stdio_destination should only be called after initialize_stdio.")

    native_engine.stdio_thread_console_set(stdin_fileno, stdout_fileno, stderr_fileno)
    try:
        yield
    finally:
        native_engine.stdio_thread_console_clear()


def stdio_destination_use_color(use_color: bool) -> None:
    """Sets a color mode for the current thread's destination.

    True or false force color to be used or not used: None causes TTY detection to decide whether
    color will be used.

    NB: This method is independent from either `stdio_destination` or `initialize_stdio` because
    we cannot decide whether to use color for a particular destination until it is open AND we have
    parsed options for the relevant connection.
    """
    native_engine.stdio_thread_console_color_mode_set(use_color)


@contextmanager
def _python_logging_setup(
    level: LogLevel, log_levels_by_target: dict[str, LogLevel], *, print_stacktrace: bool
) -> Iterator[None]:
    """Installs a root Python logger that routes all logging through a Rust logger."""

    def trace_fn(self, message, *args, **kwargs):
        if self.isEnabledFor(LogLevel.TRACE.level):
            self._log(LogLevel.TRACE.level, message, *args, **kwargs)

    logging.Logger.trace = trace_fn  # type: ignore[attr-defined]
    logger = logging.getLogger(None)

    def clear_logging_handlers():
        handlers = tuple(logger.handlers)
        for handler in handlers:
            logger.removeHandler(handler)
        return handlers

    def set_logging_handlers(handlers):
        for handler in handlers:
            logger.addHandler(handler)

    # Remove existing handlers, and restore them afterward.
    handlers = clear_logging_handlers()
    try:
        # This routes warnings through our loggers instead of straight to raw stderr.
        logging.captureWarnings(True)
        handler = _NativeHandler()
        exc_formatter = _ExceptionFormatter(level, print_stacktrace=print_stacktrace)
        handler.setFormatter(exc_formatter)
        logger.addHandler(handler)
        level.set_level_for(logger)

        for key, level in log_levels_by_target.items():
            level.set_level_for(logging.getLogger(key))

        if logger.isEnabledFor(LogLevel.TRACE.level):
            http.client.HTTPConnection.debuglevel = 1
            requests_logger = logging.getLogger("requests.packages.urllib3")
            LogLevel.TRACE.set_level_for(requests_logger)
            requests_logger.propagate = True

        yield
    finally:
        clear_logging_handlers()
        set_logging_handlers(handlers)


@contextmanager
def initialize_stdio(global_bootstrap_options: OptionValueContainer) -> Iterator[None]:
    """Mutates sys.std* and logging to route stdio for a Pants process to thread local destinations.

    In this context, `sys.std*` and logging handlers will route through Rust code that uses
    thread-local information to decide whether to write to a file, or to stdio file handles.

    To control the stdio destination set by this method, use the `stdio_destination` context manager.

    This is called in two different processes:
    * PantsRunner, after it has determined that LocalPantsRunner will be running in process, and
      immediately before setting a `stdio_destination` for the remainder of the run.
    * PantsDaemon, immediately on startup. The process will then default to sending stdio to the log
      until client connections arrive, at which point `stdio_destination` is used per-connection.
    """
    with initialize_stdio_raw(
        global_bootstrap_options.level,
        global_bootstrap_options.log_show_rust_3rdparty,
        global_bootstrap_options.show_log_target,
        _get_log_levels_by_target(global_bootstrap_options),
        global_bootstrap_options.print_stacktrace,
        global_bootstrap_options.ignore_warnings,
        global_bootstrap_options.pants_workdir,
    ):
        yield


@contextmanager
def initialize_stdio_raw(
    global_level: LogLevel,
    log_show_rust_3rdparty: bool,
    show_target: bool,
    log_levels_by_target: dict[str, LogLevel],
    print_stacktrace: bool,
    ignore_warnings: list[str],
    pants_workdir: str,
) -> Iterator[None]:
    literal_filters = []
    regex_filters = []
    for filt in ignore_warnings:
        if filt.startswith("$regex$"):
            regex_filters.append(strip_prefix(filt, "$regex$"))
        else:
            literal_filters.append(filt)

    # Set the pants log destination.
    log_path = str(pants_log_path(PurePath(pants_workdir)))
    safe_mkdir_for(log_path)

    # Initialize thread-local stdio, and replace sys.std* with proxies.
    original_stdin, original_stdout, original_stderr = sys.stdin, sys.stdout, sys.stderr
    try:
        raw_stdin, sys.stdout, sys.stderr = native_engine.stdio_initialize(
            global_level.level,
            log_show_rust_3rdparty,
            show_target,
            {k: v.level for k, v in log_levels_by_target.items()},
            tuple(literal_filters),
            tuple(regex_filters),
            log_path,
        )
        sys.stdin = TextIOWrapper(
            BufferedReader(raw_stdin),
            # NB: We set the default encoding explicitly to bypass logic in the TextIOWrapper
            # constructor that would poke the underlying file (which is not valid until a
            # `stdio_destination` is set).
            encoding=locale.getpreferredencoding(False),
        )

        sys.__stdin__, sys.__stdout__, sys.__stderr__ = sys.stdin, sys.stdout, sys.stderr  # type: ignore[misc,assignment]
        # Install a Python logger that will route through the Rust logger.
        with _python_logging_setup(
            global_level, log_levels_by_target, print_stacktrace=print_stacktrace
        ):
            yield
    finally:
        sys.stdin, sys.stdout, sys.stderr = original_stdin, original_stdout, original_stderr
        sys.__stdin__, sys.__stdout__, sys.__stderr__ = sys.stdin, sys.stdout, sys.stderr  # type: ignore[misc,assignment]


def pants_log_path(workdir: PurePath) -> PurePath:
    """Given the path of the workdir, returns the `pants.log` path."""
    return workdir / "pants.log"


def _get_log_levels_by_target(
    global_bootstrap_options: OptionValueContainer,
) -> dict[str, LogLevel]:
    raw_levels = global_bootstrap_options.log_levels_by_target
    levels: dict[str, LogLevel] = {}
    for key, value in raw_levels.items():
        if not isinstance(key, str):
            raise ValueError(
                f"Keys for log_domain_levels must be strings, but was given the key: {key} with type {type(key)}."
            )
        if not isinstance(value, str):
            raise ValueError(
                f"Values for log_domain_levels must be strings, but was given the value: {value} with type {type(value)}."
            )
        log_level = LogLevel[value.upper()]
        levels[key] = log_level
    return levels

1	# Copyright 2018 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	1✔
5
6	import http.client	1✔
7	import locale	1✔
8	import logging	1✔
9	import sys	1✔
10	from collections.abc import Iterator	1✔
11	from contextlib import contextmanager	1✔
12	from io import BufferedReader, TextIOWrapper	1✔
13	from logging import Formatter, Handler, LogRecord	1✔
14	from pathlib import PurePath	1✔
15
16	import pants.util.logging as pants_logging	1✔
17	from pants.engine.internals import native_engine	1✔
18	from pants.option.option_value_container import OptionValueContainer	1✔
19	from pants.util.dirutil import safe_mkdir_for	1✔
20	from pants.util.docutil import doc_url	1✔
21	from pants.util.logging import LogLevel	1✔
22	from pants.util.strutil import strip_prefix	1✔
23
24	# Although logging supports the WARN level, its not documented and could conceivably be yanked.
25	# Since pants has supported 'warn' since inception, leave the 'warn' choice as-is but explicitly
26	# setup a 'WARN' logging level name that maps to 'WARNING'.
27	logging.addLevelName(logging.WARNING, "WARN")	1✔
28	logging.addLevelName(pants_logging.TRACE, "TRACE")	1✔
29
30
31	class _NativeHandler(Handler):	1✔
32	"""This class is installed as a Python logging module handler (using the logging.addHandler
33	method) and proxies logs to the Rust logging infrastructure."""
34
35	def emit(self, record: LogRecord) -> None:	1✔
UNCOV 36	native_engine.write_log(self.format(record), record.levelno, record.name)	×
37
38	def flush(self) -> None:	1✔
39	native_engine.flush_log()	×
40
41
42	class _ExceptionFormatter(Formatter):	1✔
43	"""Possibly render the stacktrace and possibly give debug hints, based on global options."""
44
45	def __init__(self, level: LogLevel, *, print_stacktrace: bool) -> None:	1✔
UNCOV 46	super().__init__(None)	×
UNCOV 47	self.level = level	×
UNCOV 48	self.print_stacktrace = print_stacktrace	×
49
50	def formatException(self, exc_info):	1✔
51	stacktrace = super().formatException(exc_info) if self.print_stacktrace else ""	×
52
53	debug_instructions = []	×
54	if not self.print_stacktrace:	×
55	debug_instructions.append("--print-stacktrace for more error details")	×
56	if self.level not in {LogLevel.DEBUG, LogLevel.TRACE}:	×
57	debug_instructions.append("-ldebug for more logs")	×
58	debug_instructions = (	×
59	f"Use {' and/or '.join(debug_instructions)}. " if debug_instructions else ""
60	)
61
62	return (	×
63	f"{stacktrace}\n\n{debug_instructions}\nSee {doc_url('docs/using-pants/troubleshooting-common-issues')} for common "
64	f"issues.\nConsider reaching out for help: {doc_url('community/getting-help')}\n"
65	)
66
67
68	@contextmanager	1✔
69	def stdio_destination(stdin_fileno: int, stdout_fileno: int, stderr_fileno: int) -> Iterator[None]:	1✔
70	"""Sets a destination for both logging and stdio: must be called after `initialize_stdio`.
71
72	After `initialize_stdio` and outside of this contextmanager, the default stdio destination is
73	the pants.log. But inside of this block, all engine "tasks"/@rules that are spawned will have
74	thread/task-local state that directs their IO to the given destination. When the contextmanager
75	exits all tasks will be restored to the default destination (regardless of whether they have
76	completed).
77	"""
UNCOV 78	if not logging.getLogger(None).handlers:	×
79	raise AssertionError("stdio_destination should only be called after initialize_stdio.")	×
80
UNCOV 81	native_engine.stdio_thread_console_set(stdin_fileno, stdout_fileno, stderr_fileno)	×
UNCOV 82	try:	×
UNCOV 83	yield	×
84	finally:
UNCOV 85	native_engine.stdio_thread_console_clear()	×
86
87
88	def stdio_destination_use_color(use_color: bool) -> None:	1✔
89	"""Sets a color mode for the current thread's destination.
90
91	True or false force color to be used or not used: None causes TTY detection to decide whether
92	color will be used.
93
94	NB: This method is independent from either `stdio_destination` or `initialize_stdio` because
95	we cannot decide whether to use color for a particular destination until it is open AND we have
96	parsed options for the relevant connection.
97	"""
98	native_engine.stdio_thread_console_color_mode_set(use_color)	×
99
100
101	@contextmanager	1✔
102	def _python_logging_setup(	1✔
103	level: LogLevel, log_levels_by_target: dict[str, LogLevel], *, print_stacktrace: bool
104	) -> Iterator[None]:
105	"""Installs a root Python logger that routes all logging through a Rust logger."""
106
UNCOV 107	def trace_fn(self, message, args, *kwargs):	×
108	if self.isEnabledFor(LogLevel.TRACE.level):	×
109	self._log(LogLevel.TRACE.level, message, args, *kwargs)	×
110
UNCOV 111	logging.Logger.trace = trace_fn # type: ignore[attr-defined]	×
UNCOV 112	logger = logging.getLogger(None)	×
113
UNCOV 114	def clear_logging_handlers():	×
UNCOV 115	handlers = tuple(logger.handlers)	×
UNCOV 116	for handler in handlers:	×
UNCOV 117	logger.removeHandler(handler)	×
UNCOV 118	return handlers	×
119
UNCOV 120	def set_logging_handlers(handlers):	×
UNCOV 121	for handler in handlers:	×
UNCOV 122	logger.addHandler(handler)	×
123
124	# Remove existing handlers, and restore them afterward.
UNCOV 125	handlers = clear_logging_handlers()	×
UNCOV 126	try:	×
127	# This routes warnings through our loggers instead of straight to raw stderr.
UNCOV 128	logging.captureWarnings(True)	×
UNCOV 129	handler = _NativeHandler()	×
UNCOV 130	exc_formatter = _ExceptionFormatter(level, print_stacktrace=print_stacktrace)	×
UNCOV 131	handler.setFormatter(exc_formatter)	×
UNCOV 132	logger.addHandler(handler)	×
UNCOV 133	level.set_level_for(logger)	×
134
UNCOV 135	for key, level in log_levels_by_target.items():	×
UNCOV 136	level.set_level_for(logging.getLogger(key))	×
137
UNCOV 138	if logger.isEnabledFor(LogLevel.TRACE.level):	×
139	http.client.HTTPConnection.debuglevel = 1	×
140	requests_logger = logging.getLogger("requests.packages.urllib3")	×
141	LogLevel.TRACE.set_level_for(requests_logger)	×
142	requests_logger.propagate = True	×
143
UNCOV 144	yield	×
145	finally:
UNCOV 146	clear_logging_handlers()	×
UNCOV 147	set_logging_handlers(handlers)	×
148
149
150	@contextmanager	1✔
151	def initialize_stdio(global_bootstrap_options: OptionValueContainer) -> Iterator[None]:	1✔
152	"""Mutates sys.std* and logging to route stdio for a Pants process to thread local destinations.
153
154	In this context, `sys.std*` and logging handlers will route through Rust code that uses
155	thread-local information to decide whether to write to a file, or to stdio file handles.
156
157	To control the stdio destination set by this method, use the `stdio_destination` context manager.
158
159	This is called in two different processes:
160	* PantsRunner, after it has determined that LocalPantsRunner will be running in process, and
161	immediately before setting a `stdio_destination` for the remainder of the run.
162	* PantsDaemon, immediately on startup. The process will then default to sending stdio to the log
163	until client connections arrive, at which point `stdio_destination` is used per-connection.
164	"""
UNCOV 165	with initialize_stdio_raw(	×
166	global_bootstrap_options.level,
167	global_bootstrap_options.log_show_rust_3rdparty,
168	global_bootstrap_options.show_log_target,
169	_get_log_levels_by_target(global_bootstrap_options),
170	global_bootstrap_options.print_stacktrace,
171	global_bootstrap_options.ignore_warnings,
172	global_bootstrap_options.pants_workdir,
173	):
UNCOV 174	yield	×
175
176
177	@contextmanager	1✔
178	def initialize_stdio_raw(	1✔
179	global_level: LogLevel,
180	log_show_rust_3rdparty: bool,
181	show_target: bool,
182	log_levels_by_target: dict[str, LogLevel],
183	print_stacktrace: bool,
184	ignore_warnings: list[str],
185	pants_workdir: str,
186	) -> Iterator[None]:
UNCOV 187	literal_filters = []	×
UNCOV 188	regex_filters = []	×
UNCOV 189	for filt in ignore_warnings:	×
190	if filt.startswith("$regex$"):	×
191	regex_filters.append(strip_prefix(filt, "$regex$"))	×
192	else:
193	literal_filters.append(filt)	×
194
195	# Set the pants log destination.
UNCOV 196	log_path = str(pants_log_path(PurePath(pants_workdir)))	×
UNCOV 197	safe_mkdir_for(log_path)	×
198
199	# Initialize thread-local stdio, and replace sys.std* with proxies.
UNCOV 200	original_stdin, original_stdout, original_stderr = sys.stdin, sys.stdout, sys.stderr	×
UNCOV 201	try:	×
UNCOV 202	raw_stdin, sys.stdout, sys.stderr = native_engine.stdio_initialize(	×
203	global_level.level,
204	log_show_rust_3rdparty,
205	show_target,
206	{k: v.level for k, v in log_levels_by_target.items()},
207	tuple(literal_filters),
208	tuple(regex_filters),
209	log_path,
210	)
UNCOV 211	sys.stdin = TextIOWrapper(	×
212	BufferedReader(raw_stdin),
213	# NB: We set the default encoding explicitly to bypass logic in the TextIOWrapper
214	# constructor that would poke the underlying file (which is not valid until a
215	# `stdio_destination` is set).
216	encoding=locale.getpreferredencoding(False),
217	)
218
UNCOV 219	sys.__stdin__, sys.__stdout__, sys.__stderr__ = sys.stdin, sys.stdout, sys.stderr # type: ignore[misc,assignment]	×
220	# Install a Python logger that will route through the Rust logger.
UNCOV 221	with _python_logging_setup(	×
222	global_level, log_levels_by_target, print_stacktrace=print_stacktrace
223	):
UNCOV 224	yield	×
225	finally:
UNCOV 226	sys.stdin, sys.stdout, sys.stderr = original_stdin, original_stdout, original_stderr	×
UNCOV 227	sys.__stdin__, sys.__stdout__, sys.__stderr__ = sys.stdin, sys.stdout, sys.stderr # type: ignore[misc,assignment]	×
228
229
230	def pants_log_path(workdir: PurePath) -> PurePath:	1✔
231	"""Given the path of the workdir, returns the `pants.log` path."""
UNCOV 232	return workdir / "pants.log"	×
233
234
235	def _get_log_levels_by_target(	1✔
236	global_bootstrap_options: OptionValueContainer,
237	) -> dict[str, LogLevel]:
UNCOV 238	raw_levels = global_bootstrap_options.log_levels_by_target	×
UNCOV 239	levels: dict[str, LogLevel] = {}	×
UNCOV 240	for key, value in raw_levels.items():	×
UNCOV 241	if not isinstance(key, str):	×
242	raise ValueError(	×
243	f"Keys for log_domain_levels must be strings, but was given the key: {key} with type {type(key)}."
244	)
UNCOV 245	if not isinstance(value, str):	×
246	raise ValueError(	×
247	f"Values for log_domain_levels must be strings, but was given the value: {value} with type {type(value)}."
248	)
UNCOV 249	log_level = LogLevel[value.upper()]	×
UNCOV 250	levels[key] = log_level	×
UNCOV 251	return levels	×

pantsbuild / pants / 19015773527

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous