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

26.42

/src/python/pants/option/registrar.py

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

from __future__ import annotations

import copy
import inspect
import logging
import typing
from collections.abc import Iterator, Mapping
from dataclasses import dataclass
from enum import Enum
from typing import Any

from pants.base.deprecated import validate_deprecation_semver
from pants.option.custom_types import (
    DictValueComponent,
    ListValueComponent,
    UnsetBool,
    dir_option,
    file_option,
    shell_str,
    target_option,
)
from pants.option.errors import (
    BooleanConversionError,
    DefaultMemberValueType,
    DefaultValueType,
    HelpType,
    InvalidKwarg,
    InvalidKwargNonGlobalScope,
    InvalidMemberType,
    MemberTypeNotAllowed,
    NoOptionNames,
    OptionAlreadyRegistered,
    OptionNameDoubleDash,
    ParseError,
    PassthroughType,
    RegistrationError,
)
from pants.option.native_options import parse_dest
from pants.option.option_types import OptionInfo
from pants.option.ranked_value import RankedValue
from pants.option.scope import GLOBAL_SCOPE

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class OptionValueHistory:
    ranked_values: tuple[RankedValue, ...]

    @property
    def final_value(self) -> RankedValue:
        return self.ranked_values[-1]


class OptionRegistrar:
    """Holds information about registered options."""

    @staticmethod
    def is_bool(kwargs: Mapping[str, Any]) -> bool:
        type_arg = kwargs.get("type")
        if type_arg is None:
            return False
        if type_arg is bool:
            return True
        try:
            return typing.get_type_hints(type_arg).get("return") is bool
        except TypeError:
            return False

    @staticmethod
    def ensure_bool(val: bool | str) -> bool:
        if isinstance(val, bool):
            return val
        if isinstance(val, str):
            s = val.lower()
            if s == "true":
                return True
            if s == "false":
                return False
            raise BooleanConversionError(f'Got "{val}". Expected "True" or "False".')
        raise BooleanConversionError(f"Got {val}. Expected True or False.")

    @classmethod
    def _invert(cls, s: bool | str | None) -> bool | None:
        if s is None:
            return None
        b = cls.ensure_bool(s)
        return not b

    def __init__(self, scope: str) -> None:
        """Create an OptionRegistrar instance.

        :param scope: the scope this registrar acts for.
        """
        self._scope = scope

        # All option args registered with this registrar.  Used to prevent conflicts.
        self._known_args: set[str] = set()

        # List of (args, kwargs) registration pairs, exactly as captured at registration time.
        self._option_registrations: list[tuple[tuple[str, ...], dict[str, Any]]] = []

        # Map of dest -> history.
        self._history: dict[str, OptionValueHistory] = {}

    @property
    def scope(self) -> str:
        return self._scope

    @property
    def known_scoped_args(self) -> frozenset[str]:
        prefix = f"{self.scope}-" if self.scope != GLOBAL_SCOPE else ""
        return frozenset(f"--{prefix}{arg.lstrip('--')}" for arg in self._known_args)

    def option_registrations_iter(self) -> Iterator[OptionInfo]:
        """Returns an iterator over the normalized registration arguments of each option in this
        registrar.

        Useful for generating help and other documentation.

        Each yielded item is an OptionInfo containing the args and kwargs as passed to register(),
        except that kwargs will be normalized to always have 'dest' and 'default' explicitly set.
        """

        def normalize_kwargs(orig_args, orig_kwargs):
            nkwargs = copy.copy(orig_kwargs)
            dest = parse_dest(OptionInfo(orig_args, nkwargs))
            nkwargs["dest"] = dest
            if "default" not in nkwargs:
                type_arg = nkwargs.get("type", str)
                member_type = nkwargs.get("member_type", str)
                default_val = self.to_value_type(nkwargs.get("default"), type_arg, member_type)
                if isinstance(default_val, (ListValueComponent, DictValueComponent)):
                    default_val = default_val.val
                nkwargs["default"] = default_val
            return nkwargs

        # Yield our directly-registered options.
        for args, kwargs in self._option_registrations:
            normalized_kwargs = normalize_kwargs(args, kwargs)
            yield OptionInfo(args, normalized_kwargs)

    def register(self, *args, **kwargs) -> None:
        """Register an option."""
        self._validate(args, kwargs)

        if self.is_bool(kwargs):
            default = kwargs.get("default")
            if default is None:
                # Unless a tri-state bool is explicitly opted into with the `UnsetBool` default value,
                # boolean options always have an implicit default of False. We make that explicit here.
                kwargs["default"] = False
            elif default is UnsetBool:
                kwargs["default"] = None

        # Record the args. We'll do the underlying parsing on-demand.
        self._option_registrations.append((args, kwargs))

        # Look for direct conflicts.
        for arg in args:
            if arg in self._known_args:
                raise OptionAlreadyRegistered(self.scope, arg)
        self._known_args.update(args)

    _allowed_registration_kwargs = {
        "type",
        "member_type",
        "choices",
        "dest",
        "default",
        "default_help_repr",
        "metavar",
        "help",
        "advanced",
        "fingerprint",
        "removal_version",
        "removal_hint",
        "deprecation_start_version",
        "fromfile",
        "mutually_exclusive_group",
        "daemon",
        "passthrough",
        "environment_aware",
    }

    _allowed_member_types = {
        str,
        int,
        float,
        dict,
        dir_option,
        file_option,
        target_option,
        shell_str,
    }

    def _validate(self, args, kwargs) -> None:
        """Validate option registration arguments."""

        def error(
            exception_type: type[RegistrationError],
            arg_name: str | None = None,
            **msg_kwargs,
        ) -> None:
            if arg_name is None:
                arg_name = args[0] if args else "<unknown>"
            raise exception_type(self.scope, arg_name, **msg_kwargs)

        if not args:
            error(NoOptionNames)
        # Validate args.
        for arg in args:
            # We ban short args like `-x`, except for special casing the global option `-l`.
            if not arg.startswith("--") and not (self.scope == GLOBAL_SCOPE and arg == "-l"):
                error(OptionNameDoubleDash, arg_name=arg)

        # Validate kwargs.
        type_arg = kwargs.get("type", str)
        if "member_type" in kwargs and type_arg != list:
            error(MemberTypeNotAllowed, type_=type_arg.__name__)
        member_type = kwargs.get("member_type", str)
        is_enum = inspect.isclass(member_type) and issubclass(member_type, Enum)
        if not is_enum and member_type not in self._allowed_member_types:
            error(InvalidMemberType, member_type=member_type.__name__)

        help_arg = kwargs.get("help")
        if help_arg is not None and not isinstance(help_arg, str):
            error(HelpType, help_type=type(help_arg).__name__)

        # check type of default value
        default_value = kwargs.get("default")
        if default_value is not None:
            if isinstance(default_value, str) and type_arg != str:
                # attempt to parse default value, for correctness.
                # custom function types may implement their own validation
                default_value = self.to_value_type(default_value, type_arg, member_type)
                if hasattr(default_value, "val"):
                    default_value = default_value.val

                # fall through to type check, to verify that custom types returned a value of correct type

            if (
                isinstance(type_arg, type)
                and not isinstance(default_value, type_arg)
                and not (issubclass(type_arg, bool) and default_value == UnsetBool)
            ):
                error(
                    DefaultValueType,
                    option_type=type_arg.__name__,
                    default_value=kwargs["default"],
                    value_type=type(default_value).__name__,
                )

            # verify list member types (this is not done by the custom list value type)
            if type_arg == list:
                for member_val in default_value:
                    if not isinstance(member_type, type):
                        # defer value validation to custom type
                        member_type(member_val)

                    elif not isinstance(member_val, member_type):
                        error(
                            DefaultMemberValueType,
                            member_type=member_type.__name__,
                            member_value=member_val,
                            value_type=type(member_val).__name__,
                        )

        if (
            "passthrough" in kwargs
            and kwargs["passthrough"]
            and (type_arg != list or member_type not in (shell_str, str))
        ):
            error(PassthroughType)

        for kwarg in kwargs:
            if kwarg not in self._allowed_registration_kwargs:
                error(InvalidKwarg, kwarg=kwarg)

            # Ensure `daemon=True` can't be passed on non-global scopes.
            if kwarg == "daemon" and self.scope != GLOBAL_SCOPE:
                error(InvalidKwargNonGlobalScope, kwarg=kwarg)

        removal_version = kwargs.get("removal_version")
        if removal_version is not None:
            validate_deprecation_semver(removal_version, "removal version")

    @classmethod
    def to_value_type(cls, val_str, type_arg, member_type):
        """Convert a string to a value of the option's type."""
        if val_str is None:
            return None
        if type_arg == bool:
            return cls.ensure_bool(val_str)
        try:
            if type_arg == list:
                return ListValueComponent.create(val_str, member_type=member_type)
            if type_arg == dict:
                return DictValueComponent.create(val_str)
            return type_arg(val_str)
        except (TypeError, ValueError) as e:
            if issubclass(type_arg, Enum):
                choices = ", ".join(f"{choice.value}" for choice in type_arg)
                raise ParseError(f"Invalid choice '{val_str}'. Choose from: {choices}")
            raise ParseError(
                f"Error applying type '{type_arg.__name__}' to option value '{val_str}': {e}"
            )

    def __str__(self) -> str:
        return f"OptionRegistrar({self.scope})"

1	# Copyright 2014 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 copy	1✔
7	import inspect	1✔
8	import logging	1✔
9	import typing	1✔
10	from collections.abc import Iterator, Mapping	1✔
11	from dataclasses import dataclass	1✔
12	from enum import Enum	1✔
13	from typing import Any	1✔
14
15	from pants.base.deprecated import validate_deprecation_semver	1✔
16	from pants.option.custom_types import (	1✔
17	DictValueComponent,
18	ListValueComponent,
19	UnsetBool,
20	dir_option,
21	file_option,
22	shell_str,
23	target_option,
24	)
25	from pants.option.errors import (	1✔
26	BooleanConversionError,
27	DefaultMemberValueType,
28	DefaultValueType,
29	HelpType,
30	InvalidKwarg,
31	InvalidKwargNonGlobalScope,
32	InvalidMemberType,
33	MemberTypeNotAllowed,
34	NoOptionNames,
35	OptionAlreadyRegistered,
36	OptionNameDoubleDash,
37	ParseError,
38	PassthroughType,
39	RegistrationError,
40	)
41	from pants.option.native_options import parse_dest	1✔
42	from pants.option.option_types import OptionInfo	1✔
43	from pants.option.ranked_value import RankedValue	1✔
44	from pants.option.scope import GLOBAL_SCOPE	1✔
45
46	logger = logging.getLogger(__name__)	1✔
47
48
49	@dataclass(frozen=True)	1✔
50	class OptionValueHistory:	1✔
51	ranked_values: tuple[RankedValue, ...]	1✔
52
53	@property	1✔
54	def final_value(self) -> RankedValue:	1✔
UNCOV 55	return self.ranked_values[-1]	×
56
57
58	class OptionRegistrar:	1✔
59	"""Holds information about registered options."""
60
61	@staticmethod	1✔
62	def is_bool(kwargs: Mapping[str, Any]) -> bool:	1✔
UNCOV 63	type_arg = kwargs.get("type")	×
UNCOV 64	if type_arg is None:	×
UNCOV 65	return False	×
UNCOV 66	if type_arg is bool:	×
UNCOV 67	return True	×
UNCOV 68	try:	×
UNCOV 69	return typing.get_type_hints(type_arg).get("return") is bool	×
70	except TypeError:	×
71	return False	×
72
73	@staticmethod	1✔
74	def ensure_bool(val: bool \| str) -> bool:	1✔
UNCOV 75	if isinstance(val, bool):	×
76	return val	×
UNCOV 77	if isinstance(val, str):	×
UNCOV 78	s = val.lower()	×
UNCOV 79	if s == "true":	×
80	return True	×
UNCOV 81	if s == "false":	×
82	return False	×
UNCOV 83	raise BooleanConversionError(f'Got "{val}". Expected "True" or "False".')	×
84	raise BooleanConversionError(f"Got {val}. Expected True or False.")	×
85
86	@classmethod	1✔
87	def _invert(cls, s: bool \| str \| None) -> bool \| None:	1✔
88	if s is None:	×
89	return None	×
90	b = cls.ensure_bool(s)	×
91	return not b	×
92
93	def __init__(self, scope: str) -> None:	1✔
94	"""Create an OptionRegistrar instance.
95
96	:param scope: the scope this registrar acts for.
97	"""
UNCOV 98	self._scope = scope	×
99
100	# All option args registered with this registrar. Used to prevent conflicts.
UNCOV 101	self._known_args: set[str] = set()	×
102
103	# List of (args, kwargs) registration pairs, exactly as captured at registration time.
UNCOV 104	self._option_registrations: list[tuple[tuple[str, ...], dict[str, Any]]] = []	×
105
106	# Map of dest -> history.
UNCOV 107	self._history: dict[str, OptionValueHistory] = {}	×
108
109	@property	1✔
110	def scope(self) -> str:	1✔
UNCOV 111	return self._scope	×
112
113	@property	1✔
114	def known_scoped_args(self) -> frozenset[str]:	1✔
UNCOV 115	prefix = f"{self.scope}-" if self.scope != GLOBAL_SCOPE else ""	×
UNCOV 116	return frozenset(f"--{prefix}{arg.lstrip('--')}" for arg in self._known_args)	×
117
118	def option_registrations_iter(self) -> Iterator[OptionInfo]:	1✔
119	"""Returns an iterator over the normalized registration arguments of each option in this
120	registrar.
121
122	Useful for generating help and other documentation.
123
124	Each yielded item is an OptionInfo containing the args and kwargs as passed to register(),
125	except that kwargs will be normalized to always have 'dest' and 'default' explicitly set.
126	"""
127
UNCOV 128	def normalize_kwargs(orig_args, orig_kwargs):	×
UNCOV 129	nkwargs = copy.copy(orig_kwargs)	×
UNCOV 130	dest = parse_dest(OptionInfo(orig_args, nkwargs))	×
UNCOV 131	nkwargs["dest"] = dest	×
UNCOV 132	if "default" not in nkwargs:	×
UNCOV 133	type_arg = nkwargs.get("type", str)	×
UNCOV 134	member_type = nkwargs.get("member_type", str)	×
UNCOV 135	default_val = self.to_value_type(nkwargs.get("default"), type_arg, member_type)	×
UNCOV 136	if isinstance(default_val, (ListValueComponent, DictValueComponent)):	×
137	default_val = default_val.val	×
UNCOV 138	nkwargs["default"] = default_val	×
UNCOV 139	return nkwargs	×
140
141	# Yield our directly-registered options.
UNCOV 142	for args, kwargs in self._option_registrations:	×
UNCOV 143	normalized_kwargs = normalize_kwargs(args, kwargs)	×
UNCOV 144	yield OptionInfo(args, normalized_kwargs)	×
145
146	def register(self, args, *kwargs) -> None:	1✔
147	"""Register an option."""
UNCOV 148	self._validate(args, kwargs)	×
149
UNCOV 150	if self.is_bool(kwargs):	×
UNCOV 151	default = kwargs.get("default")	×
UNCOV 152	if default is None:	×
153	# Unless a tri-state bool is explicitly opted into with the `UnsetBool` default value,
154	# boolean options always have an implicit default of False. We make that explicit here.
UNCOV 155	kwargs["default"] = False	×
UNCOV 156	elif default is UnsetBool:	×
UNCOV 157	kwargs["default"] = None	×
158
159	# Record the args. We'll do the underlying parsing on-demand.
UNCOV 160	self._option_registrations.append((args, kwargs))	×
161
162	# Look for direct conflicts.
UNCOV 163	for arg in args:	×
UNCOV 164	if arg in self._known_args:	×
UNCOV 165	raise OptionAlreadyRegistered(self.scope, arg)	×
UNCOV 166	self._known_args.update(args)	×
167
168	_allowed_registration_kwargs = {	1✔
169	"type",
170	"member_type",
171	"choices",
172	"dest",
173	"default",
174	"default_help_repr",
175	"metavar",
176	"help",
177	"advanced",
178	"fingerprint",
179	"removal_version",
180	"removal_hint",
181	"deprecation_start_version",
182	"fromfile",
183	"mutually_exclusive_group",
184	"daemon",
185	"passthrough",
186	"environment_aware",
187	}
188
189	_allowed_member_types = {	1✔
190	str,
191	int,
192	float,
193	dict,
194	dir_option,
195	file_option,
196	target_option,
197	shell_str,
198	}
199
200	def _validate(self, args, kwargs) -> None:	1✔
201	"""Validate option registration arguments."""
202
UNCOV 203	def error(	×
204	exception_type: type[RegistrationError],
205	arg_name: str \| None = None,
206	**msg_kwargs,
207	) -> None:
UNCOV 208	if arg_name is None:	×
UNCOV 209	arg_name = args[0] if args else "<unknown>"	×
UNCOV 210	raise exception_type(self.scope, arg_name, **msg_kwargs)	×
211
UNCOV 212	if not args:	×
UNCOV 213	error(NoOptionNames)	×
214	# Validate args.
UNCOV 215	for arg in args:	×
216	# We ban short args like `-x`, except for special casing the global option `-l`.
UNCOV 217	if not arg.startswith("--") and not (self.scope == GLOBAL_SCOPE and arg == "-l"):	×
UNCOV 218	error(OptionNameDoubleDash, arg_name=arg)	×
219
220	# Validate kwargs.
UNCOV 221	type_arg = kwargs.get("type", str)	×
UNCOV 222	if "member_type" in kwargs and type_arg != list:	×
UNCOV 223	error(MemberTypeNotAllowed, type_=type_arg.__name__)	×
UNCOV 224	member_type = kwargs.get("member_type", str)	×
UNCOV 225	is_enum = inspect.isclass(member_type) and issubclass(member_type, Enum)	×
UNCOV 226	if not is_enum and member_type not in self._allowed_member_types:	×
UNCOV 227	error(InvalidMemberType, member_type=member_type.__name__)	×
228
UNCOV 229	help_arg = kwargs.get("help")	×
UNCOV 230	if help_arg is not None and not isinstance(help_arg, str):	×
UNCOV 231	error(HelpType, help_type=type(help_arg).__name__)	×
232
233	# check type of default value
UNCOV 234	default_value = kwargs.get("default")	×
UNCOV 235	if default_value is not None:	×
UNCOV 236	if isinstance(default_value, str) and type_arg != str:	×
237	# attempt to parse default value, for correctness.
238	# custom function types may implement their own validation
UNCOV 239	default_value = self.to_value_type(default_value, type_arg, member_type)	×
UNCOV 240	if hasattr(default_value, "val"):	×
UNCOV 241	default_value = default_value.val	×
242
243	# fall through to type check, to verify that custom types returned a value of correct type
244
UNCOV 245	if (	×
246	isinstance(type_arg, type)
247	and not isinstance(default_value, type_arg)
248	and not (issubclass(type_arg, bool) and default_value == UnsetBool)
249	):
UNCOV 250	error(	×
251	DefaultValueType,
252	option_type=type_arg.__name__,
253	default_value=kwargs["default"],
254	value_type=type(default_value).__name__,
255	)
256
257	# verify list member types (this is not done by the custom list value type)
UNCOV 258	if type_arg == list:	×
UNCOV 259	for member_val in default_value:	×
UNCOV 260	if not isinstance(member_type, type):	×
261	# defer value validation to custom type
UNCOV 262	member_type(member_val)	×
263
UNCOV 264	elif not isinstance(member_val, member_type):	×
UNCOV 265	error(	×
266	DefaultMemberValueType,
267	member_type=member_type.__name__,
268	member_value=member_val,
269	value_type=type(member_val).__name__,
270	)
271
UNCOV 272	if (	×
273	"passthrough" in kwargs
274	and kwargs["passthrough"]
275	and (type_arg != list or member_type not in (shell_str, str))
276	):
277	error(PassthroughType)	×
278
UNCOV 279	for kwarg in kwargs:	×
UNCOV 280	if kwarg not in self._allowed_registration_kwargs:	×
UNCOV 281	error(InvalidKwarg, kwarg=kwarg)	×
282
283	# Ensure `daemon=True` can't be passed on non-global scopes.
UNCOV 284	if kwarg == "daemon" and self.scope != GLOBAL_SCOPE:	×
285	error(InvalidKwargNonGlobalScope, kwarg=kwarg)	×
286
UNCOV 287	removal_version = kwargs.get("removal_version")	×
UNCOV 288	if removal_version is not None:	×
UNCOV 289	validate_deprecation_semver(removal_version, "removal version")	×
290
291	@classmethod	1✔
292	def to_value_type(cls, val_str, type_arg, member_type):	1✔
293	"""Convert a string to a value of the option's type."""
UNCOV 294	if val_str is None:	×
UNCOV 295	return None	×
UNCOV 296	if type_arg == bool:	×
UNCOV 297	return cls.ensure_bool(val_str)	×
UNCOV 298	try:	×
UNCOV 299	if type_arg == list:	×
UNCOV 300	return ListValueComponent.create(val_str, member_type=member_type)	×
UNCOV 301	if type_arg == dict:	×
UNCOV 302	return DictValueComponent.create(val_str)	×
UNCOV 303	return type_arg(val_str)	×
UNCOV 304	except (TypeError, ValueError) as e:	×
305	if issubclass(type_arg, Enum):	×
306	choices = ", ".join(f"{choice.value}" for choice in type_arg)	×
307	raise ParseError(f"Invalid choice '{val_str}'. Choose from: {choices}")	×
308	raise ParseError(	×
309	f"Error applying type '{type_arg.__name__}' to option value '{val_str}': {e}"
310	)
311
312	def __str__(self) -> str:	1✔
313	return f"OptionRegistrar({self.scope})"	×

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