18252174847

Committed 05 Oct 2025 01:36AM UTC coverage: 43.382% (-36.9%) from 80.261%

Build # 18252174847

Build Type

push

github

Committed by

web-flow

Commit Message

run tests on mac arm (#22717)

Just doing the minimal to pull forward the x86_64 pattern.

ref #20993

Run Details

25776 of 59416 relevant lines covered (43.38%)

1.3 hits per line

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

76.07

/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	3✔
5
6	import http.client	3✔
7	import locale	3✔
8	import logging	3✔
9	import sys	3✔
10	from collections.abc import Iterator	3✔
11	from contextlib import contextmanager	3✔
12	from io import BufferedReader, TextIOWrapper	3✔
13	from logging import Formatter, Handler, LogRecord	3✔
14	from pathlib import PurePath	3✔
15
16	import pants.util.logging as pants_logging	3✔
17	from pants.engine.internals import native_engine	3✔
18	from pants.option.option_value_container import OptionValueContainer	3✔
19	from pants.util.dirutil import safe_mkdir_for	3✔
20	from pants.util.docutil import doc_url	3✔
21	from pants.util.logging import LogLevel	3✔
22	from pants.util.strutil import strip_prefix	3✔
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")	3✔
28	logging.addLevelName(pants_logging.TRACE, "TRACE")	3✔
29
30
31	class _NativeHandler(Handler):	3✔
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:	3✔
36	native_engine.write_log(self.format(record), record.levelno, record.name)	×
37
38	def flush(self) -> None:	3✔
39	native_engine.flush_log()	×
40
41
42	class _ExceptionFormatter(Formatter):	3✔
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:	3✔
46	super().__init__(None)	3✔
47	self.level = level	3✔
48	self.print_stacktrace = print_stacktrace	3✔
49
50	def formatException(self, exc_info):	3✔
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	3✔
69	def stdio_destination(stdin_fileno: int, stdout_fileno: int, stderr_fileno: int) -> Iterator[None]:	3✔
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	"""
78	if not logging.getLogger(None).handlers:	3✔
79	raise AssertionError("stdio_destination should only be called after initialize_stdio.")	×
80
81	native_engine.stdio_thread_console_set(stdin_fileno, stdout_fileno, stderr_fileno)	3✔
82	try:	3✔
83	yield	3✔
84	finally:
85	native_engine.stdio_thread_console_clear()	3✔
86
87
88	def stdio_destination_use_color(use_color: bool) -> None:	3✔
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	3✔
102	def _python_logging_setup(	3✔
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
107	def trace_fn(self, message, args, *kwargs):	3✔
108	if self.isEnabledFor(LogLevel.TRACE.level):	×
109	self._log(LogLevel.TRACE.level, message, args, *kwargs)	×
110
111	logging.Logger.trace = trace_fn # type: ignore[attr-defined]	3✔
112	logger = logging.getLogger(None)	3✔
113
114	def clear_logging_handlers():	3✔
115	handlers = tuple(logger.handlers)	3✔
116	for handler in handlers:	3✔
117	logger.removeHandler(handler)	3✔
118	return handlers	3✔
119
120	def set_logging_handlers(handlers):	3✔
121	for handler in handlers:	3✔
122	logger.addHandler(handler)	3✔
123
124	# Remove existing handlers, and restore them afterward.
125	handlers = clear_logging_handlers()	3✔
126	try:	3✔
127	# This routes warnings through our loggers instead of straight to raw stderr.
128	logging.captureWarnings(True)	3✔
129	handler = _NativeHandler()	3✔
130	exc_formatter = _ExceptionFormatter(level, print_stacktrace=print_stacktrace)	3✔
131	handler.setFormatter(exc_formatter)	3✔
132	logger.addHandler(handler)	3✔
133	level.set_level_for(logger)	3✔
134
135	for key, level in log_levels_by_target.items():	3✔
136	level.set_level_for(logging.getLogger(key))	×
137
138	if logger.isEnabledFor(LogLevel.TRACE.level):	3✔
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
144	yield	3✔
145	finally:
146	clear_logging_handlers()	3✔
147	set_logging_handlers(handlers)	3✔
148
149
150	@contextmanager	3✔
151	def initialize_stdio(global_bootstrap_options: OptionValueContainer) -> Iterator[None]:	3✔
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	"""
165	with initialize_stdio_raw(	3✔
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	):
174	yield	3✔
175
176
177	@contextmanager	3✔
178	def initialize_stdio_raw(	3✔
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]:
187	literal_filters = []	3✔
188	regex_filters = []	3✔
189	for filt in ignore_warnings:	3✔
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.
196	log_path = str(pants_log_path(PurePath(pants_workdir)))	3✔
197	safe_mkdir_for(log_path)	3✔
198
199	# Initialize thread-local stdio, and replace sys.std* with proxies.
200	original_stdin, original_stdout, original_stderr = sys.stdin, sys.stdout, sys.stderr	3✔
201	try:	3✔
202	raw_stdin, sys.stdout, sys.stderr = native_engine.stdio_initialize(	3✔
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	)
211	sys.stdin = TextIOWrapper(	3✔
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
219	sys.__stdin__, sys.__stdout__, sys.__stderr__ = sys.stdin, sys.stdout, sys.stderr # type: ignore[misc,assignment]	3✔
220	# Install a Python logger that will route through the Rust logger.
221	with _python_logging_setup(	3✔
222	global_level, log_levels_by_target, print_stacktrace=print_stacktrace
223	):
224	yield	3✔
225	finally:
226	sys.stdin, sys.stdout, sys.stderr = original_stdin, original_stdout, original_stderr	3✔
227	sys.__stdin__, sys.__stdout__, sys.__stderr__ = sys.stdin, sys.stdout, sys.stderr # type: ignore[misc,assignment]	3✔
228
229
230	def pants_log_path(workdir: PurePath) -> PurePath:	3✔
231	"""Given the path of the workdir, returns the `pants.log` path."""
232	return workdir / "pants.log"	3✔
233
234
235	def _get_log_levels_by_target(	3✔
236	global_bootstrap_options: OptionValueContainer,
237	) -> dict[str, LogLevel]:
238	raw_levels = global_bootstrap_options.log_levels_by_target	3✔
239	levels: dict[str, LogLevel] = {}	3✔
240	for key, value in raw_levels.items():	3✔
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	)
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	)
249	log_level = LogLevel[value.upper()]	×
250	levels[key] = log_level	×
251	return levels	3✔

pantsbuild / pants / 18252174847

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