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

32.92

/src/python/pants/option/options.py

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

from __future__ import annotations

import dataclasses
import logging
from collections import defaultdict
from collections.abc import Iterable, Mapping, Sequence
from typing import Any

from pants.base.deprecated import warn_or_error
from pants.engine.fs import FileContent
from pants.engine.internals.native_engine import PyGoalInfo, PyPantsCommand
from pants.option.errors import (
    ConfigValidationError,
    MutuallyExclusiveOptionError,
    UnknownFlagsError,
)
from pants.option.native_options import NativeOptionParser
from pants.option.option_util import is_list_option
from pants.option.option_value_container import OptionValueContainer, OptionValueContainerBuilder
from pants.option.ranked_value import Rank, RankedValue
from pants.option.registrar import OptionRegistrar
from pants.option.scope import GLOBAL_SCOPE, GLOBAL_SCOPE_CONFIG_SECTION, ScopeInfo
from pants.util.memo import memoized_method
from pants.util.ordered_set import FrozenOrderedSet, OrderedSet
from pants.util.strutil import softwrap

logger = logging.getLogger(__name__)


class Options:
    """The outward-facing API for interacting with options.

    Supports option registration and fetching option values.

    Examples:

    The value in global scope of option '--foo-bar' (registered in global scope) will be selected
    in the following order:
      - The value of the --foo-bar flag in global scope.
      - The value of the PANTS_GLOBAL_FOO_BAR environment variable.
      - The value of the PANTS_FOO_BAR environment variable.
      - The value of the foo_bar key in the [GLOBAL] section of pants.toml.
      - The hard-coded value provided at registration time.
      - None.

    The value in scope 'compile.java' of option '--foo-bar' (registered in global scope) will be
    selected in the following order:
      - The value of the --foo-bar flag in scope 'compile.java'.
      - The value of the --foo-bar flag in scope 'compile'.
      - The value of the --foo-bar flag in global scope.
      - The value of the PANTS_COMPILE_JAVA_FOO_BAR environment variable.
      - The value of the PANTS_COMPILE_FOO_BAR environment variable.
      - The value of the PANTS_GLOBAL_FOO_BAR environment variable.
      - The value of the PANTS_FOO_BAR environment variable.
      - The value of the foo_bar key in the [compile.java] section of pants.toml.
      - The value of the foo_bar key in the [compile] section of pants.toml.
      - The value of the foo_bar key in the [GLOBAL] section of pants.toml.
      - The hard-coded value provided at registration time.
      - None.

    The value in scope 'compile.java' of option '--foo-bar' (registered in scope 'compile') will be
    selected in the following order:
      - The value of the --foo-bar flag in scope 'compile.java'.
      - The value of the --foo-bar flag in scope 'compile'.
      - The value of the PANTS_COMPILE_JAVA_FOO_BAR environment variable.
      - The value of the PANTS_COMPILE_FOO_BAR environment variable.
      - The value of the foo_bar key in the [compile.java] section of pants.toml.
      - The value of the foo_bar key in the [compile] section of pants.toml.
      - The value of the foo_bar key in the [GLOBAL] section of pants.toml
        (because of automatic config file fallback to that section).
      - The hard-coded value provided at registration time.
      - None.
    """

    class DuplicateScopeError(Exception):
        """More than one registration occurred for the same scope."""

    class AmbiguousPassthroughError(Exception):
        """More than one goal was passed along with passthrough args."""

    @classmethod
    def complete_scopes(cls, scope_infos: Iterable[ScopeInfo]) -> FrozenOrderedSet[ScopeInfo]:
        """Expand a set of scopes to include scopes they deprecate.

        Also validates that scopes do not collide.
        """
        ret: OrderedSet[ScopeInfo] = OrderedSet()
        original_scopes: dict[str, ScopeInfo] = {}
        for si in sorted(scope_infos, key=lambda _si: _si.scope):
            if si.scope in original_scopes:
                raise cls.DuplicateScopeError(
                    softwrap(
                        f"""
                        Scope `{si.scope}` claimed by {si}, was also claimed
                        by {original_scopes[si.scope]}.
                        """
                    )
                )
            original_scopes[si.scope] = si
            ret.add(si)
            if si.deprecated_scope:
                ret.add(dataclasses.replace(si, scope=si.deprecated_scope))
                original_scopes[si.deprecated_scope] = si
        return FrozenOrderedSet(ret)

    @classmethod
    def create(
        cls,
        *,
        args: Sequence[str],
        env: Mapping[str, str],
        config_sources: Sequence[FileContent] | None,
        known_scope_infos: Sequence[ScopeInfo],
        allow_unknown_options: bool = False,
        allow_pantsrc: bool = True,
        include_derivation: bool = False,
    ) -> Options:
        """Create an Options instance.

        :param args: a list of cmd-line args; defaults to `sys.argv` if None is supplied.
        :param env: a dict of environment variables.
        :param config_sources: sources of config data.
        :param known_scope_infos: ScopeInfos for all scopes that may be encountered.
        :param allow_unknown_options: Whether to ignore or error on unknown cmd-line flags.
        :param allow_pantsrc: Whether to read config from local .rc files. Typically
          disabled in tests, for hermeticity.
        :param include_derivation: Whether to gather option value derivation information.
        """
        # We need registrars for all the intermediate scopes, so inherited option values
        # can propagate through them.
        complete_known_scope_infos = cls.complete_scopes(known_scope_infos)

        registrar_by_scope = {
            si.scope: OptionRegistrar(si.scope) for si in complete_known_scope_infos
        }
        known_scope_to_info = {s.scope: s for s in complete_known_scope_infos}
        known_scope_to_flags = {
            scope: registrar.known_scoped_args for scope, registrar in registrar_by_scope.items()
        }
        known_goals = tuple(
            PyGoalInfo(si.scope, si.is_builtin, si.is_auxiliary, si.scope_aliases)
            for si in known_scope_infos
            if si.is_goal
        )

        native_parser = NativeOptionParser(
            args[1:],  # The native parser expects args without the sys.argv[0] binary name.
            env,
            config_sources=config_sources,
            allow_pantsrc=allow_pantsrc,
            include_derivation=include_derivation,
            known_scopes_to_flags=known_scope_to_flags,
            known_goals=known_goals,
        )

        command = native_parser.get_command()
        if command.passthru() and len(command.goals()) > 1:
            raise cls.AmbiguousPassthroughError(
                softwrap(
                    f"""
                    Specifying multiple goals (in this case: {command.goals()})
                    along with passthrough args (args after `--`) is ambiguous.

                    Try either specifying only a single goal, or passing the passthrough args
                    directly to the relevant consumer via its associated flags.
                    """
                )
            )

        return cls(
            command=command,
            registrar_by_scope=registrar_by_scope,
            native_parser=native_parser,
            known_scope_to_info=known_scope_to_info,
            allow_unknown_options=allow_unknown_options,
        )

    def __init__(
        self,
        command: PyPantsCommand,
        registrar_by_scope: dict[str, OptionRegistrar],
        native_parser: NativeOptionParser,
        known_scope_to_info: dict[str, ScopeInfo],
        allow_unknown_options: bool = False,
    ) -> None:
        """The low-level constructor for an Options instance.

        Dependents should use `Options.create` instead.
        """
        self._command = command
        self._registrar_by_scope = registrar_by_scope
        self._native_parser = native_parser
        self._known_scope_to_info = known_scope_to_info
        self._allow_unknown_options = allow_unknown_options

    @property
    def native_parser(self) -> NativeOptionParser:
        return self._native_parser

    @property
    def specs(self) -> list[str]:
        """The specifications to operate on, e.g. the target addresses and the file names.

        :API: public
        """
        return self._command.specs()

    @property
    def builtin_or_auxiliary_goal(self) -> str | None:
        """The requested builtin or auxiliary goal, if any.

        :API: public
        """
        return self._command.builtin_or_auxiliary_goal()

    @property
    def goals(self) -> list[str]:
        """The requested goals, in the order specified on the cmd line.

        :API: public
        """
        return self._command.goals()

    @property
    def unknown_goals(self) -> list[str]:
        """The requested goals without implementation, in the order specified on the cmd line.

        :API: public
        """
        return self._command.unknown_goals()

    @property
    def known_scope_to_info(self) -> dict[str, ScopeInfo]:
        return self._known_scope_to_info

    @property
    def known_scope_to_scoped_args(self) -> dict[str, frozenset[str]]:
        return {
            scope: registrar.known_scoped_args
            for scope, registrar in self._registrar_by_scope.items()
        }

    def verify_configs(self) -> None:
        """Verify all loaded configs have correct scopes and options."""

        section_to_valid_options = {}
        for scope in self.known_scope_to_info:
            section = GLOBAL_SCOPE_CONFIG_SECTION if scope == GLOBAL_SCOPE else scope
            section_to_valid_options[section] = set(self.for_scope(scope, check_deprecations=False))

        error_log = self.native_parser.validate_config(section_to_valid_options)
        if error_log:
            for error in error_log:
                logger.error(error)
            raise ConfigValidationError(
                softwrap(
                    """
                    Invalid config entries detected. See log for details on which entries to update
                    or remove.

                    (Specify --no-verify-config to disable this check.)
                    """
                )
            )

    def verify_args(self):
        # Consume all known args, and see if any are left.
        # This will have the side-effect of precomputing (and memoizing) options for all scopes.
        for scope in self.known_scope_to_info:
            self.for_scope(scope)
        # We implement some global help flags, such as `-h`, `--help`, '-v', `--version`,
        # as scope aliases (so `--help` is an alias for `help` and so on).
        # There aren't consumed by the native parser, since they aren't registered as options,
        # so we must account for them.
        scope_aliases_that_look_like_flags = set()
        for si in self.known_scope_to_info.values():
            scope_aliases_that_look_like_flags.update(
                sa for sa in si.scope_aliases if sa.startswith("-")
            )

        for scope, flags in self._native_parser.get_unconsumed_flags().items():
            flags = tuple(flag for flag in flags if flag not in scope_aliases_that_look_like_flags)
            if flags:
                # We may have unconsumed flags in multiple positional contexts, but our
                # error handling expects just one, so pick the first one. After the user
                # fixes that error we will show the next scope.
                raise UnknownFlagsError(flags, scope)

    def is_known_scope(self, scope: str) -> bool:
        """Whether the given scope is known by this instance.

        :API: public
        """
        return scope in self._known_scope_to_info

    def register(self, scope: str, *args, **kwargs) -> None:
        """Register an option in the given scope."""
        self.get_registrar(scope).register(*args, **kwargs)
        deprecated_scope = self.known_scope_to_info[scope].deprecated_scope
        if deprecated_scope:
            self.get_registrar(deprecated_scope).register(*args, **kwargs)

    def get_registrar(self, scope: str) -> OptionRegistrar:
        """Returns the registrar for the given scope, so code can register on it directly.

        :param scope: The scope to retrieve the registrar for.
        :return: The registrar for the given scope.
        :raises pants.option.errors.ConfigValidationError: if the scope is not known.
        """
        try:
            return self._registrar_by_scope[scope]
        except KeyError:
            raise ConfigValidationError(f"No such options scope: {scope}")

    def _check_and_apply_deprecations(self, scope, values):
        """Checks whether a ScopeInfo has options specified in a deprecated scope.

        There are two related cases here. Either:
          1) The ScopeInfo has an associated deprecated_scope that was replaced with a non-deprecated
             scope, meaning that the options temporarily live in two locations.
          2) The entire ScopeInfo is deprecated (as in the case of deprecated SubsystemDependencies),
             meaning that the options live in one location.

        In the first case, this method has the side effect of merging options values from deprecated
        scopes into the given values.
        """
        si = self.known_scope_to_info[scope]

        # If this Scope is itself deprecated, report that.
        if si.removal_version:
            explicit_keys = self.for_scope(scope, check_deprecations=False).get_explicit_keys()
            if explicit_keys:
                warn_or_error(
                    removal_version=si.removal_version,
                    entity=f"scope {scope}",
                    hint=si.removal_hint,
                )

        # Check if we're the new name of a deprecated scope, and clone values from that scope.
        # Note that deprecated_scope and scope share the same Subsystem class, so deprecated_scope's
        # Subsystem has a deprecated_options_scope equal to deprecated_scope. Therefore we must
        # check that scope != deprecated_scope to prevent infinite recursion.
        deprecated_scope = si.deprecated_scope
        if deprecated_scope is not None and scope != deprecated_scope:
            # Do the deprecation check only on keys that were explicitly set
            # on the deprecated scope.
            explicit_keys = self.for_scope(
                deprecated_scope, check_deprecations=False
            ).get_explicit_keys()
            if explicit_keys:
                # Update our values with those of the deprecated scope.
                # Note that a deprecated val will take precedence over a val of equal rank.
                # This makes the code a bit neater.
                values.update(self.for_scope(deprecated_scope))

                warn_or_error(
                    removal_version=self.known_scope_to_info[
                        scope
                    ].deprecated_scope_removal_version,
                    entity=f"scope {deprecated_scope}",
                    hint=f"Use scope {scope} instead (options: {', '.join(explicit_keys)})",
                )

    # TODO: Eagerly precompute backing data for this?
    @memoized_method
    def for_scope(
        self,
        scope: str,
        check_deprecations: bool = True,
    ) -> OptionValueContainer:
        """Return the option values for the given scope.

        Values are attributes of the returned object, e.g., options.foo.
        Computed lazily per scope.

        :API: public
        :param scope: The scope to get options for.
        :param check_deprecations: Whether to check for any deprecations conditions.
        :return: An OptionValueContainer representing the option values for the given scope.
        :raises pants.option.errors.ConfigValidationError: if the scope is unknown.
        """
        builder = OptionValueContainerBuilder()
        mutex_map = defaultdict(list)
        registrar = self.get_registrar(scope)
        scope_str = "global scope" if scope == GLOBAL_SCOPE else f"scope '{scope}'"

        for option_info in registrar.option_registrations_iter():
            dest = option_info.kwargs["dest"]
            val, rank = self._native_parser.get_value(scope=scope, option_info=option_info)
            explicitly_set = rank > Rank.HARDCODED

            # If we explicitly set a deprecated but not-yet-expired option, warn about it.
            # Otherwise, raise a CodeRemovedError if the deprecation has expired.
            removal_version = option_info.kwargs.get("removal_version", None)
            if removal_version is not None:
                warn_or_error(
                    removal_version=removal_version,
                    entity=f"option '{dest}' in {scope_str}",
                    start_version=option_info.kwargs.get("deprecation_start_version", None),
                    hint=option_info.kwargs.get("removal_hint", None),
                    print_warning=explicitly_set,
                )

            # If we explicitly set the option, check for mutual exclusivity.
            if explicitly_set:
                mutex_dest = option_info.kwargs.get("mutually_exclusive_group")
                mutex_map_key = mutex_dest or dest
                mutex_map[mutex_map_key].append(dest)
                if len(mutex_map[mutex_map_key]) > 1:
                    raise MutuallyExclusiveOptionError(
                        softwrap(
                            f"""
                            Can only provide one of these mutually exclusive options in
                            {scope_str}, but multiple given:
                            {", ".join(mutex_map[mutex_map_key])}
                            """
                        )
                    )
            setattr(builder, dest, RankedValue(rank, val))

        # Check for any deprecation conditions, which are evaluated using `self._flag_matchers`.
        if check_deprecations:
            self._check_and_apply_deprecations(scope, builder)
        return builder.build()

    def get_fingerprintable_for_scope(
        self,
        scope: str,
        daemon_only: bool = False,
    ) -> list[tuple[str, type, Any]]:
        """Returns a list of fingerprintable (option name, option type, option value) pairs for the
        given scope.

        Options are fingerprintable by default, but may be registered with "fingerprint=False".

        This method also searches enclosing options scopes of `bottom_scope` to determine the set of
        fingerprintable pairs.

        :param scope: The scope to gather fingerprintable options for.
        :param daemon_only: If true, only look at daemon=True options.
        """

        pairs = []
        registrar = self.get_registrar(scope)
        # Sort the arguments, so that the fingerprint is consistent.
        for option_info in sorted(registrar.option_registrations_iter()):
            if not option_info.kwargs.get("fingerprint", True):
                continue
            if daemon_only and not option_info.kwargs.get("daemon", False):
                continue
            dest = option_info.kwargs["dest"]
            val = self.for_scope(scope)[dest]
            # If we have a list then we delegate to the fingerprinting implementation of the members.
            if is_list_option(option_info.kwargs):
                val_type = option_info.kwargs.get("member_type", str)
            else:
                val_type = option_info.kwargs.get("type", str)
            pairs.append((dest, val_type, val))
        return pairs

    def __getitem__(self, scope: str) -> OptionValueContainer:
        # TODO(John Sirois): Mainly supports use of dict<str, dict<str, str>> for mock options in tests,
        # Consider killing if tests consolidate on using TestOptions instead of the raw dicts.
        return self.for_scope(scope)

    def for_global_scope(self) -> OptionValueContainer:
        """Return the option values for the global scope.

        :API: public
        """
        return self.for_scope(GLOBAL_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 dataclasses	1✔
7	import logging	1✔
8	from collections import defaultdict	1✔
9	from collections.abc import Iterable, Mapping, Sequence	1✔
10	from typing import Any	1✔
11
12	from pants.base.deprecated import warn_or_error	1✔
13	from pants.engine.fs import FileContent	1✔
14	from pants.engine.internals.native_engine import PyGoalInfo, PyPantsCommand	1✔
15	from pants.option.errors import (	1✔
16	ConfigValidationError,
17	MutuallyExclusiveOptionError,
18	UnknownFlagsError,
19	)
20	from pants.option.native_options import NativeOptionParser	1✔
21	from pants.option.option_util import is_list_option	1✔
22	from pants.option.option_value_container import OptionValueContainer, OptionValueContainerBuilder	1✔
23	from pants.option.ranked_value import Rank, RankedValue	1✔
24	from pants.option.registrar import OptionRegistrar	1✔
25	from pants.option.scope import GLOBAL_SCOPE, GLOBAL_SCOPE_CONFIG_SECTION, ScopeInfo	1✔
26	from pants.util.memo import memoized_method	1✔
27	from pants.util.ordered_set import FrozenOrderedSet, OrderedSet	1✔
28	from pants.util.strutil import softwrap	1✔
29
30	logger = logging.getLogger(__name__)	1✔
31
32
33	class Options:	1✔
34	"""The outward-facing API for interacting with options.
35
36	Supports option registration and fetching option values.
37
38	Examples:
39
40	The value in global scope of option '--foo-bar' (registered in global scope) will be selected
41	in the following order:
42	- The value of the --foo-bar flag in global scope.
43	- The value of the PANTS_GLOBAL_FOO_BAR environment variable.
44	- The value of the PANTS_FOO_BAR environment variable.
45	- The value of the foo_bar key in the [GLOBAL] section of pants.toml.
46	- The hard-coded value provided at registration time.
47	- None.
48
49	The value in scope 'compile.java' of option '--foo-bar' (registered in global scope) will be
50	selected in the following order:
51	- The value of the --foo-bar flag in scope 'compile.java'.
52	- The value of the --foo-bar flag in scope 'compile'.
53	- The value of the --foo-bar flag in global scope.
54	- The value of the PANTS_COMPILE_JAVA_FOO_BAR environment variable.
55	- The value of the PANTS_COMPILE_FOO_BAR environment variable.
56	- The value of the PANTS_GLOBAL_FOO_BAR environment variable.
57	- The value of the PANTS_FOO_BAR environment variable.
58	- The value of the foo_bar key in the [compile.java] section of pants.toml.
59	- The value of the foo_bar key in the [compile] section of pants.toml.
60	- The value of the foo_bar key in the [GLOBAL] section of pants.toml.
61	- The hard-coded value provided at registration time.
62	- None.
63
64	The value in scope 'compile.java' of option '--foo-bar' (registered in scope 'compile') will be
65	selected in the following order:
66	- The value of the --foo-bar flag in scope 'compile.java'.
67	- The value of the --foo-bar flag in scope 'compile'.
68	- The value of the PANTS_COMPILE_JAVA_FOO_BAR environment variable.
69	- The value of the PANTS_COMPILE_FOO_BAR environment variable.
70	- The value of the foo_bar key in the [compile.java] section of pants.toml.
71	- The value of the foo_bar key in the [compile] section of pants.toml.
72	- The value of the foo_bar key in the [GLOBAL] section of pants.toml
73	(because of automatic config file fallback to that section).
74	- The hard-coded value provided at registration time.
75	- None.
76	"""
77
78	class DuplicateScopeError(Exception):	1✔
79	"""More than one registration occurred for the same scope."""
80
81	class AmbiguousPassthroughError(Exception):	1✔
82	"""More than one goal was passed along with passthrough args."""
83
84	@classmethod	1✔
85	def complete_scopes(cls, scope_infos: Iterable[ScopeInfo]) -> FrozenOrderedSet[ScopeInfo]:	1✔
86	"""Expand a set of scopes to include scopes they deprecate.
87
88	Also validates that scopes do not collide.
89	"""
UNCOV 90	ret: OrderedSet[ScopeInfo] = OrderedSet()	×
UNCOV 91	original_scopes: dict[str, ScopeInfo] = {}	×
UNCOV 92	for si in sorted(scope_infos, key=lambda _si: _si.scope):	×
UNCOV 93	if si.scope in original_scopes:	×
UNCOV 94	raise cls.DuplicateScopeError(	×
95	softwrap(
96	f"""
97	Scope `{si.scope}` claimed by {si}, was also claimed
98	by {original_scopes[si.scope]}.
99	"""
100	)
101	)
UNCOV 102	original_scopes[si.scope] = si	×
UNCOV 103	ret.add(si)	×
UNCOV 104	if si.deprecated_scope:	×
UNCOV 105	ret.add(dataclasses.replace(si, scope=si.deprecated_scope))	×
UNCOV 106	original_scopes[si.deprecated_scope] = si	×
UNCOV 107	return FrozenOrderedSet(ret)	×
108
109	@classmethod	1✔
110	def create(	1✔
111	cls,
112	*,
113	args: Sequence[str],
114	env: Mapping[str, str],
115	config_sources: Sequence[FileContent] \| None,
116	known_scope_infos: Sequence[ScopeInfo],
117	allow_unknown_options: bool = False,
118	allow_pantsrc: bool = True,
119	include_derivation: bool = False,
120	) -> Options:
121	"""Create an Options instance.
122
123	:param args: a list of cmd-line args; defaults to `sys.argv` if None is supplied.
124	:param env: a dict of environment variables.
125	:param config_sources: sources of config data.
126	:param known_scope_infos: ScopeInfos for all scopes that may be encountered.
127	:param allow_unknown_options: Whether to ignore or error on unknown cmd-line flags.
128	:param allow_pantsrc: Whether to read config from local .rc files. Typically
129	disabled in tests, for hermeticity.
130	:param include_derivation: Whether to gather option value derivation information.
131	"""
132	# We need registrars for all the intermediate scopes, so inherited option values
133	# can propagate through them.
UNCOV 134	complete_known_scope_infos = cls.complete_scopes(known_scope_infos)	×
135
UNCOV 136	registrar_by_scope = {	×
137	si.scope: OptionRegistrar(si.scope) for si in complete_known_scope_infos
138	}
UNCOV 139	known_scope_to_info = {s.scope: s for s in complete_known_scope_infos}	×
UNCOV 140	known_scope_to_flags = {	×
141	scope: registrar.known_scoped_args for scope, registrar in registrar_by_scope.items()
142	}
UNCOV 143	known_goals = tuple(	×
144	PyGoalInfo(si.scope, si.is_builtin, si.is_auxiliary, si.scope_aliases)
145	for si in known_scope_infos
146	if si.is_goal
147	)
148
UNCOV 149	native_parser = NativeOptionParser(	×
150	args[1:], # The native parser expects args without the sys.argv[0] binary name.
151	env,
152	config_sources=config_sources,
153	allow_pantsrc=allow_pantsrc,
154	include_derivation=include_derivation,
155	known_scopes_to_flags=known_scope_to_flags,
156	known_goals=known_goals,
157	)
158
UNCOV 159	command = native_parser.get_command()	×
UNCOV 160	if command.passthru() and len(command.goals()) > 1:	×
UNCOV 161	raise cls.AmbiguousPassthroughError(	×
162	softwrap(
163	f"""
164	Specifying multiple goals (in this case: {command.goals()})
165	along with passthrough args (args after `--`) is ambiguous.
166
167	Try either specifying only a single goal, or passing the passthrough args
168	directly to the relevant consumer via its associated flags.
169	"""
170	)
171	)
172
UNCOV 173	return cls(	×
174	command=command,
175	registrar_by_scope=registrar_by_scope,
176	native_parser=native_parser,
177	known_scope_to_info=known_scope_to_info,
178	allow_unknown_options=allow_unknown_options,
179	)
180
181	def __init__(	1✔
182	self,
183	command: PyPantsCommand,
184	registrar_by_scope: dict[str, OptionRegistrar],
185	native_parser: NativeOptionParser,
186	known_scope_to_info: dict[str, ScopeInfo],
187	allow_unknown_options: bool = False,
188	) -> None:
189	"""The low-level constructor for an Options instance.
190
191	Dependents should use `Options.create` instead.
192	"""
UNCOV 193	self._command = command	×
UNCOV 194	self._registrar_by_scope = registrar_by_scope	×
UNCOV 195	self._native_parser = native_parser	×
UNCOV 196	self._known_scope_to_info = known_scope_to_info	×
UNCOV 197	self._allow_unknown_options = allow_unknown_options	×
198
199	@property	1✔
200	def native_parser(self) -> NativeOptionParser:	1✔
UNCOV 201	return self._native_parser	×
202
203	@property	1✔
204	def specs(self) -> list[str]:	1✔
205	"""The specifications to operate on, e.g. the target addresses and the file names.
206
207	:API: public
208	"""
UNCOV 209	return self._command.specs()	×
210
211	@property	1✔
212	def builtin_or_auxiliary_goal(self) -> str \| None:	1✔
213	"""The requested builtin or auxiliary goal, if any.
214
215	:API: public
216	"""
217	return self._command.builtin_or_auxiliary_goal()	×
218
219	@property	1✔
220	def goals(self) -> list[str]:	1✔
221	"""The requested goals, in the order specified on the cmd line.
222
223	:API: public
224	"""
UNCOV 225	return self._command.goals()	×
226
227	@property	1✔
228	def unknown_goals(self) -> list[str]:	1✔
229	"""The requested goals without implementation, in the order specified on the cmd line.
230
231	:API: public
232	"""
233	return self._command.unknown_goals()	×
234
235	@property	1✔
236	def known_scope_to_info(self) -> dict[str, ScopeInfo]:	1✔
UNCOV 237	return self._known_scope_to_info	×
238
239	@property	1✔
240	def known_scope_to_scoped_args(self) -> dict[str, frozenset[str]]:	1✔
241	return {	×
242	scope: registrar.known_scoped_args
243	for scope, registrar in self._registrar_by_scope.items()
244	}
245
246	def verify_configs(self) -> None:	1✔
247	"""Verify all loaded configs have correct scopes and options."""
248
249	section_to_valid_options = {}	×
250	for scope in self.known_scope_to_info:	×
251	section = GLOBAL_SCOPE_CONFIG_SECTION if scope == GLOBAL_SCOPE else scope	×
252	section_to_valid_options[section] = set(self.for_scope(scope, check_deprecations=False))	×
253
254	error_log = self.native_parser.validate_config(section_to_valid_options)	×
255	if error_log:	×
256	for error in error_log:	×
257	logger.error(error)	×
258	raise ConfigValidationError(	×
259	softwrap(
260	"""
261	Invalid config entries detected. See log for details on which entries to update
262	or remove.
263
264	(Specify --no-verify-config to disable this check.)
265	"""
266	)
267	)
268
269	def verify_args(self):	1✔
270	# Consume all known args, and see if any are left.
271	# This will have the side-effect of precomputing (and memoizing) options for all scopes.
UNCOV 272	for scope in self.known_scope_to_info:	×
UNCOV 273	self.for_scope(scope)	×
274	# We implement some global help flags, such as `-h`, `--help`, '-v', `--version`,
275	# as scope aliases (so `--help` is an alias for `help` and so on).
276	# There aren't consumed by the native parser, since they aren't registered as options,
277	# so we must account for them.
UNCOV 278	scope_aliases_that_look_like_flags = set()	×
UNCOV 279	for si in self.known_scope_to_info.values():	×
UNCOV 280	scope_aliases_that_look_like_flags.update(	×
281	sa for sa in si.scope_aliases if sa.startswith("-")
282	)
283
UNCOV 284	for scope, flags in self._native_parser.get_unconsumed_flags().items():	×
UNCOV 285	flags = tuple(flag for flag in flags if flag not in scope_aliases_that_look_like_flags)	×
UNCOV 286	if flags:	×
287	# We may have unconsumed flags in multiple positional contexts, but our
288	# error handling expects just one, so pick the first one. After the user
289	# fixes that error we will show the next scope.
UNCOV 290	raise UnknownFlagsError(flags, scope)	×
291
292	def is_known_scope(self, scope: str) -> bool:	1✔
293	"""Whether the given scope is known by this instance.
294
295	:API: public
296	"""
UNCOV 297	return scope in self._known_scope_to_info	×
298
299	def register(self, scope: str, args, *kwargs) -> None:	1✔
300	"""Register an option in the given scope."""
UNCOV 301	self.get_registrar(scope).register(args, *kwargs)	×
UNCOV 302	deprecated_scope = self.known_scope_to_info[scope].deprecated_scope	×
UNCOV 303	if deprecated_scope:	×
UNCOV 304	self.get_registrar(deprecated_scope).register(args, *kwargs)	×
305
306	def get_registrar(self, scope: str) -> OptionRegistrar:	1✔
307	"""Returns the registrar for the given scope, so code can register on it directly.
308
309	:param scope: The scope to retrieve the registrar for.
310	:return: The registrar for the given scope.
311	:raises pants.option.errors.ConfigValidationError: if the scope is not known.
312	"""
UNCOV 313	try:	×
UNCOV 314	return self._registrar_by_scope[scope]	×
315	except KeyError:	×
316	raise ConfigValidationError(f"No such options scope: {scope}")	×
317
318	def _check_and_apply_deprecations(self, scope, values):	1✔
319	"""Checks whether a ScopeInfo has options specified in a deprecated scope.
320
321	There are two related cases here. Either:
322	1) The ScopeInfo has an associated deprecated_scope that was replaced with a non-deprecated
323	scope, meaning that the options temporarily live in two locations.
324	2) The entire ScopeInfo is deprecated (as in the case of deprecated SubsystemDependencies),
325	meaning that the options live in one location.
326
327	In the first case, this method has the side effect of merging options values from deprecated
328	scopes into the given values.
329	"""
UNCOV 330	si = self.known_scope_to_info[scope]	×
331
332	# If this Scope is itself deprecated, report that.
UNCOV 333	if si.removal_version:	×
334	explicit_keys = self.for_scope(scope, check_deprecations=False).get_explicit_keys()	×
335	if explicit_keys:	×
336	warn_or_error(	×
337	removal_version=si.removal_version,
338	entity=f"scope {scope}",
339	hint=si.removal_hint,
340	)
341
342	# Check if we're the new name of a deprecated scope, and clone values from that scope.
343	# Note that deprecated_scope and scope share the same Subsystem class, so deprecated_scope's
344	# Subsystem has a deprecated_options_scope equal to deprecated_scope. Therefore we must
345	# check that scope != deprecated_scope to prevent infinite recursion.
UNCOV 346	deprecated_scope = si.deprecated_scope	×
UNCOV 347	if deprecated_scope is not None and scope != deprecated_scope:	×
348	# Do the deprecation check only on keys that were explicitly set
349	# on the deprecated scope.
UNCOV 350	explicit_keys = self.for_scope(	×
351	deprecated_scope, check_deprecations=False
352	).get_explicit_keys()
UNCOV 353	if explicit_keys:	×
354	# Update our values with those of the deprecated scope.
355	# Note that a deprecated val will take precedence over a val of equal rank.
356	# This makes the code a bit neater.
UNCOV 357	values.update(self.for_scope(deprecated_scope))	×
358
UNCOV 359	warn_or_error(	×
360	removal_version=self.known_scope_to_info[
361	scope
362	].deprecated_scope_removal_version,
363	entity=f"scope {deprecated_scope}",
364	hint=f"Use scope {scope} instead (options: {', '.join(explicit_keys)})",
365	)
366
367	# TODO: Eagerly precompute backing data for this?
368	@memoized_method	1✔
369	def for_scope(	1✔
370	self,
371	scope: str,
372	check_deprecations: bool = True,
373	) -> OptionValueContainer:
374	"""Return the option values for the given scope.
375
376	Values are attributes of the returned object, e.g., options.foo.
377	Computed lazily per scope.
378
379	:API: public
380	:param scope: The scope to get options for.
381	:param check_deprecations: Whether to check for any deprecations conditions.
382	:return: An OptionValueContainer representing the option values for the given scope.
383	:raises pants.option.errors.ConfigValidationError: if the scope is unknown.
384	"""
UNCOV 385	builder = OptionValueContainerBuilder()	×
UNCOV 386	mutex_map = defaultdict(list)	×
UNCOV 387	registrar = self.get_registrar(scope)	×
UNCOV 388	scope_str = "global scope" if scope == GLOBAL_SCOPE else f"scope '{scope}'"	×
389
UNCOV 390	for option_info in registrar.option_registrations_iter():	×
UNCOV 391	dest = option_info.kwargs["dest"]	×
UNCOV 392	val, rank = self._native_parser.get_value(scope=scope, option_info=option_info)	×
UNCOV 393	explicitly_set = rank > Rank.HARDCODED	×
394
395	# If we explicitly set a deprecated but not-yet-expired option, warn about it.
396	# Otherwise, raise a CodeRemovedError if the deprecation has expired.
UNCOV 397	removal_version = option_info.kwargs.get("removal_version", None)	×
UNCOV 398	if removal_version is not None:	×
UNCOV 399	warn_or_error(	×
400	removal_version=removal_version,
401	entity=f"option '{dest}' in {scope_str}",
402	start_version=option_info.kwargs.get("deprecation_start_version", None),
403	hint=option_info.kwargs.get("removal_hint", None),
404	print_warning=explicitly_set,
405	)
406
407	# If we explicitly set the option, check for mutual exclusivity.
UNCOV 408	if explicitly_set:	×
UNCOV 409	mutex_dest = option_info.kwargs.get("mutually_exclusive_group")	×
UNCOV 410	mutex_map_key = mutex_dest or dest	×
UNCOV 411	mutex_map[mutex_map_key].append(dest)	×
UNCOV 412	if len(mutex_map[mutex_map_key]) > 1:	×
UNCOV 413	raise MutuallyExclusiveOptionError(	×
414	softwrap(
415	f"""
416	Can only provide one of these mutually exclusive options in
417	{scope_str}, but multiple given:
418	{", ".join(mutex_map[mutex_map_key])}
419	"""
420	)
421	)
UNCOV 422	setattr(builder, dest, RankedValue(rank, val))	×
423
424	# Check for any deprecation conditions, which are evaluated using `self._flag_matchers`.
UNCOV 425	if check_deprecations:	×
UNCOV 426	self._check_and_apply_deprecations(scope, builder)	×
UNCOV 427	return builder.build()	×
428
429	def get_fingerprintable_for_scope(	1✔
430	self,
431	scope: str,
432	daemon_only: bool = False,
433	) -> list[tuple[str, type, Any]]:
434	"""Returns a list of fingerprintable (option name, option type, option value) pairs for the
435	given scope.
436
437	Options are fingerprintable by default, but may be registered with "fingerprint=False".
438
439	This method also searches enclosing options scopes of `bottom_scope` to determine the set of
440	fingerprintable pairs.
441
442	:param scope: The scope to gather fingerprintable options for.
443	:param daemon_only: If true, only look at daemon=True options.
444	"""
445
UNCOV 446	pairs = []	×
UNCOV 447	registrar = self.get_registrar(scope)	×
448	# Sort the arguments, so that the fingerprint is consistent.
UNCOV 449	for option_info in sorted(registrar.option_registrations_iter()):	×
UNCOV 450	if not option_info.kwargs.get("fingerprint", True):	×
UNCOV 451	continue	×
UNCOV 452	if daemon_only and not option_info.kwargs.get("daemon", False):	×
UNCOV 453	continue	×
UNCOV 454	dest = option_info.kwargs["dest"]	×
UNCOV 455	val = self.for_scope(scope)[dest]	×
456	# If we have a list then we delegate to the fingerprinting implementation of the members.
UNCOV 457	if is_list_option(option_info.kwargs):	×
UNCOV 458	val_type = option_info.kwargs.get("member_type", str)	×
459	else:
UNCOV 460	val_type = option_info.kwargs.get("type", str)	×
UNCOV 461	pairs.append((dest, val_type, val))	×
UNCOV 462	return pairs	×
463
464	def __getitem__(self, scope: str) -> OptionValueContainer:	1✔
465	# TODO(John Sirois): Mainly supports use of dict<str, dict<str, str>> for mock options in tests,
466	# Consider killing if tests consolidate on using TestOptions instead of the raw dicts.
467	return self.for_scope(scope)	×
468
469	def for_global_scope(self) -> OptionValueContainer:	1✔
470	"""Return the option values for the global scope.
471
472	:API: public
473	"""
UNCOV 474	return self.for_scope(GLOBAL_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