20632486505

Committed 01 Jan 2026 04:21AM UTC coverage: 43.231% (-37.1%) from 80.281%

Build # 20632486505

Build Type

Pull #22962

github

Committed by

web-flow

Commit Message

Merge 08d5c63b0 into f52ab6675

Pull Request Pull Request #22962: Bump the gha-deps group across 1 directory with 6 updates

Run Details

26122 of 60424 relevant lines covered (43.23%)

0.86 hits per line

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

67.92

/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	2✔
5
6	import copy	2✔
7	import inspect	2✔
8	import logging	2✔
9	import typing	2✔
10	from collections.abc import Iterator, Mapping	2✔
11	from dataclasses import dataclass	2✔
12	from enum import Enum	2✔
13	from typing import Any	2✔
14
15	from pants.base.deprecated import validate_deprecation_semver	2✔
16	from pants.option.custom_types import (	2✔
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 (	2✔
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	2✔
42	from pants.option.option_types import OptionInfo	2✔
43	from pants.option.ranked_value import RankedValue	2✔
44	from pants.option.scope import GLOBAL_SCOPE	2✔
45
46	logger = logging.getLogger(__name__)	2✔
47
48
49	@dataclass(frozen=True)	2✔
50	class OptionValueHistory:	2✔
51	ranked_values: tuple[RankedValue, ...]	2✔
52
53	@property	2✔
54	def final_value(self) -> RankedValue:	2✔
55	return self.ranked_values[-1]	×
56
57
58	class OptionRegistrar:	2✔
59	"""Holds information about registered options."""
60
61	@staticmethod	2✔
62	def is_bool(kwargs: Mapping[str, Any]) -> bool:	2✔
63	type_arg = kwargs.get("type")	2✔
64	if type_arg is None:	2✔
65	return False	×
66	if type_arg is bool:	2✔
67	return True	2✔
68	try:	2✔
69	return typing.get_type_hints(type_arg).get("return") is bool	2✔
70	except TypeError:	×
71	return False	×
72
73	@staticmethod	2✔
74	def ensure_bool(val: bool \| str) -> bool:	2✔
75	if isinstance(val, bool):	×
76	return val	×
77	if isinstance(val, str):	×
78	s = val.lower()	×
79	if s == "true":	×
80	return True	×
81	if s == "false":	×
82	return False	×
83	raise BooleanConversionError(f'Got "{val}". Expected "True" or "False".')	×
84	raise BooleanConversionError(f"Got {val}. Expected True or False.")	×
85
86	@classmethod	2✔
87	def _invert(cls, s: bool \| str \| None) -> bool \| None:	2✔
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:	2✔
94	"""Create an OptionRegistrar instance.
95
96	:param scope: the scope this registrar acts for.
97	"""
98	self._scope = scope	2✔
99
100	# All option args registered with this registrar. Used to prevent conflicts.
101	self._known_args: set[str] = set()	2✔
102
103	# List of (args, kwargs) registration pairs, exactly as captured at registration time.
104	self._option_registrations: list[tuple[tuple[str, ...], dict[str, Any]]] = []	2✔
105
106	# Map of dest -> history.
107	self._history: dict[str, OptionValueHistory] = {}	2✔
108
109	@property	2✔
110	def scope(self) -> str:	2✔
111	return self._scope	2✔
112
113	@property	2✔
114	def known_scoped_args(self) -> frozenset[str]:	2✔
115	prefix = f"{self.scope}-" if self.scope != GLOBAL_SCOPE else ""	2✔
116	return frozenset(f"--{prefix}{arg.lstrip('--')}" for arg in self._known_args)	2✔
117
118	def option_registrations_iter(self) -> Iterator[OptionInfo]:	2✔
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
128	def normalize_kwargs(orig_args, orig_kwargs):	2✔
129	nkwargs = copy.copy(orig_kwargs)	2✔
130	dest = parse_dest(OptionInfo(orig_args, nkwargs))	2✔
131	nkwargs["dest"] = dest	2✔
132	if "default" not in nkwargs:	2✔
133	type_arg = nkwargs.get("type", str)	×
134	member_type = nkwargs.get("member_type", str)	×
135	default_val = self.to_value_type(nkwargs.get("default"), type_arg, member_type)	×
136	if isinstance(default_val, (ListValueComponent, DictValueComponent)):	×
137	default_val = default_val.val	×
138	nkwargs["default"] = default_val	×
139	return nkwargs	2✔
140
141	# Yield our directly-registered options.
142	for args, kwargs in self._option_registrations:	2✔
143	normalized_kwargs = normalize_kwargs(args, kwargs)	2✔
144	yield OptionInfo(args, normalized_kwargs)	2✔
145
146	def register(self, args, *kwargs) -> None:	2✔
147	"""Register an option."""
148	self._validate(args, kwargs)	2✔
149
150	if self.is_bool(kwargs):	2✔
151	default = kwargs.get("default")	2✔
152	if default is None:	2✔
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.
155	kwargs["default"] = False	×
156	elif default is UnsetBool:	2✔
157	kwargs["default"] = None	×
158
159	# Record the args. We'll do the underlying parsing on-demand.
160	self._option_registrations.append((args, kwargs))	2✔
161
162	# Look for direct conflicts.
163	for arg in args:	2✔
164	if arg in self._known_args:	2✔
165	raise OptionAlreadyRegistered(self.scope, arg)	×
166	self._known_args.update(args)	2✔
167
168	_allowed_registration_kwargs = {	2✔
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 = {	2✔
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:	2✔
201	"""Validate option registration arguments."""
202
203	def error(	2✔
204	exception_type: type[RegistrationError],
205	arg_name: str \| None = None,
206	**msg_kwargs,
207	) -> None:
208	if arg_name is None:	×
209	arg_name = args[0] if args else "<unknown>"	×
210	raise exception_type(self.scope, arg_name, **msg_kwargs)	×
211
212	if not args:	2✔
213	error(NoOptionNames)	×
214	# Validate args.
215	for arg in args:	2✔
216	# We ban short args like `-x`, except for special casing the global option `-l`.
217	if not arg.startswith("--") and not (self.scope == GLOBAL_SCOPE and arg == "-l"):	2✔
218	error(OptionNameDoubleDash, arg_name=arg)	×
219
220	# Validate kwargs.
221	type_arg = kwargs.get("type", str)	2✔
222	if "member_type" in kwargs and type_arg != list:	2✔
223	error(MemberTypeNotAllowed, type_=type_arg.__name__)	×
224	member_type = kwargs.get("member_type", str)	2✔
225	is_enum = inspect.isclass(member_type) and issubclass(member_type, Enum)	2✔
226	if not is_enum and member_type not in self._allowed_member_types:	2✔
227	error(InvalidMemberType, member_type=member_type.__name__)	×
228
229	help_arg = kwargs.get("help")	2✔
230	if help_arg is not None and not isinstance(help_arg, str):	2✔
231	error(HelpType, help_type=type(help_arg).__name__)	×
232
233	# check type of default value
234	default_value = kwargs.get("default")	2✔
235	if default_value is not None:	2✔
236	if isinstance(default_value, str) and type_arg != str:	2✔
237	# attempt to parse default value, for correctness.
238	# custom function types may implement their own validation
239	default_value = self.to_value_type(default_value, type_arg, member_type)	2✔
240	if hasattr(default_value, "val"):	2✔
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
245	if (	2✔
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	):
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)
258	if type_arg == list:	2✔
259	for member_val in default_value:	2✔
260	if not isinstance(member_type, type):	2✔
261	# defer value validation to custom type
262	member_type(member_val)	2✔
263
264	elif not isinstance(member_val, member_type):	2✔
265	error(	×
266	DefaultMemberValueType,
267	member_type=member_type.__name__,
268	member_value=member_val,
269	value_type=type(member_val).__name__,
270	)
271
272	if (	2✔
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
279	for kwarg in kwargs:	2✔
280	if kwarg not in self._allowed_registration_kwargs:	2✔
281	error(InvalidKwarg, kwarg=kwarg)	×
282
283	# Ensure `daemon=True` can't be passed on non-global scopes.
284	if kwarg == "daemon" and self.scope != GLOBAL_SCOPE:	2✔
285	error(InvalidKwargNonGlobalScope, kwarg=kwarg)	×
286
287	removal_version = kwargs.get("removal_version")	2✔
288	if removal_version is not None:	2✔
289	validate_deprecation_semver(removal_version, "removal version")	2✔
290
291	@classmethod	2✔
292	def to_value_type(cls, val_str, type_arg, member_type):	2✔
293	"""Convert a string to a value of the option's type."""
294	if val_str is None:	2✔
295	return None	×
296	if type_arg == bool:	2✔
297	return cls.ensure_bool(val_str)	×
298	try:	2✔
299	if type_arg == list:	2✔
300	return ListValueComponent.create(val_str, member_type=member_type)	×
301	if type_arg == dict:	2✔
302	return DictValueComponent.create(val_str)	×
303	return type_arg(val_str)	2✔
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:	2✔
313	return f"OptionRegistrar({self.scope})"	×

pantsbuild / pants / 20632486505

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