19529437518

Committed 20 Nov 2025 07:44AM UTC coverage: 78.884% (-1.4%) from 80.302%

Build # 19529437518

Build Type

push

github

Committed by

web-flow

Commit Message

nfpm.native_libs: Add RPM package depends from packaged pex_binaries (#22899)

## PR Series Overview

This is the second in a series of PRs that introduces a new backend:
`pants.backend.npm.native_libs`
Initially, the backend will be available as:
`pants.backend.experimental.nfpm.native_libs`

I proposed this new backend (originally named `bindeps`) in discussion
#22396.

This backend will inspect ELF bin/lib files (like `lib*.so`) in packaged
contents (for this PR series, only in `pex_binary` targets) to identify
package dependency metadata and inject that metadata on the relevant
`nfpm_deb_package` or `nfpm_rpm_package` targets. Effectively, it will
provide an approximation of these native packager features:
- `rpm`: `rpmdeps` + `elfdeps`
- `deb`: `dh_shlibdeps` + `dpkg-shlibdeps` (These substitute
`${shlibs:Depends}` in debian control files have)

### Goal: Host-agnostic package builds

This pants backend is designed to be host-agnostic, like
[nFPM](https://nfpm.goreleaser.com/).

Native packaging tools are often restricted to a single release of a
single distro. Unlike native package builders, this new pants backend
does not use any of those distro-specific or distro-release-specific
utilities or local package databases. This new backend should be able to
run (help with building deb and rpm packages) anywhere that pants can
run (MacOS, rpm linux distros, deb linux distros, other linux distros,
docker, ...).

### Previous PRs in series

- #22873

## PR Overview

This PR adds rules in `nfpm.native_libs` to add package dependency
metadata to `nfpm_rpm_package`. The 2 new rules are:

- `inject_native_libs_dependencies_in_package_fields`:

    - An implementation of the polymorphic rule `inject_nfpm_package_fields`.
      This rule is low priority (`priority = 2`) so that in-repo plugins can
      override/augment what it injects. (See #22864)

    - Rule logic overview:
        - find any pex_binaries that will be packaged in an `nfpm_rpm_package`
   ... (continued)

Run Details

96 of 118 new or added lines in 3 files covered. (81.36%)

910 existing lines in 53 files now uncovered.

73897 of 93678 relevant lines covered (78.88%)

3.21 hits per line

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

77.02

/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	11✔
5
6	import dataclasses	11✔
7	import logging	11✔
8	from collections import defaultdict	11✔
9	from collections.abc import Iterable, Mapping, Sequence	11✔
10	from typing import Any	11✔
11
12	from pants.base.deprecated import warn_or_error	11✔
13	from pants.engine.fs import FileContent	11✔
14	from pants.engine.internals.native_engine import PyGoalInfo, PyPantsCommand	11✔
15	from pants.option.errors import (	11✔
16	ConfigValidationError,
17	MutuallyExclusiveOptionError,
18	UnknownFlagsError,
19	)
20	from pants.option.native_options import NativeOptionParser	11✔
21	from pants.option.option_util import is_list_option	11✔
22	from pants.option.option_value_container import OptionValueContainer, OptionValueContainerBuilder	11✔
23	from pants.option.ranked_value import Rank, RankedValue	11✔
24	from pants.option.registrar import OptionRegistrar	11✔
25	from pants.option.scope import GLOBAL_SCOPE, GLOBAL_SCOPE_CONFIG_SECTION, ScopeInfo	11✔
26	from pants.util.memo import memoized_method	11✔
27	from pants.util.ordered_set import FrozenOrderedSet, OrderedSet	11✔
28	from pants.util.strutil import softwrap	11✔
29
30	logger = logging.getLogger(__name__)	11✔
31
32
33	class Options:	11✔
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):	11✔
79	"""More than one registration occurred for the same scope."""
80
81	class AmbiguousPassthroughError(Exception):	11✔
82	"""More than one goal was passed along with passthrough args."""
83
84	@classmethod	11✔
85	def complete_scopes(cls, scope_infos: Iterable[ScopeInfo]) -> FrozenOrderedSet[ScopeInfo]:	11✔
86	"""Expand a set of scopes to include scopes they deprecate.
87
88	Also validates that scopes do not collide.
89	"""
90	ret: OrderedSet[ScopeInfo] = OrderedSet()	11✔
91	original_scopes: dict[str, ScopeInfo] = {}	11✔
92	for si in sorted(scope_infos, key=lambda _si: _si.scope):	11✔
93	if si.scope in original_scopes:	11✔
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	)
102	original_scopes[si.scope] = si	11✔
103	ret.add(si)	11✔
104	if si.deprecated_scope:	11✔
105	ret.add(dataclasses.replace(si, scope=si.deprecated_scope))	1✔
106	original_scopes[si.deprecated_scope] = si	1✔
107	return FrozenOrderedSet(ret)	11✔
108
109	@classmethod	11✔
110	def create(	11✔
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.
134	complete_known_scope_infos = cls.complete_scopes(known_scope_infos)	11✔
135
136	registrar_by_scope = {	11✔
137	si.scope: OptionRegistrar(si.scope) for si in complete_known_scope_infos
138	}
139	known_scope_to_info = {s.scope: s for s in complete_known_scope_infos}	11✔
140	known_scope_to_flags = {	11✔
141	scope: registrar.known_scoped_args for scope, registrar in registrar_by_scope.items()
142	}
143	known_goals = tuple(	11✔
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
149	native_parser = NativeOptionParser(	11✔
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
159	command = native_parser.get_command()	11✔
160	if command.passthru() and len(command.goals()) > 1:	11✔
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
173	return cls(	11✔
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__(	11✔
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	"""
193	self._command = command	11✔
194	self._registrar_by_scope = registrar_by_scope	11✔
195	self._native_parser = native_parser	11✔
196	self._known_scope_to_info = known_scope_to_info	11✔
197	self._allow_unknown_options = allow_unknown_options	11✔
198
199	@property	11✔
200	def native_parser(self) -> NativeOptionParser:	11✔
201	return self._native_parser	2✔
202
203	@property	11✔
204	def specs(self) -> list[str]:	11✔
205	"""The specifications to operate on, e.g. the target addresses and the file names.
206
207	:API: public
208	"""
209	return self._command.specs()	11✔
210
211	@property	11✔
212	def builtin_or_auxiliary_goal(self) -> str \| None:	11✔
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	11✔
220	def goals(self) -> list[str]:	11✔
221	"""The requested goals, in the order specified on the cmd line.
222
223	:API: public
224	"""
225	return self._command.goals()	1✔
226
227	@property	11✔
228	def unknown_goals(self) -> list[str]:	11✔
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	11✔
236	def known_scope_to_info(self) -> dict[str, ScopeInfo]:	11✔
237	return self._known_scope_to_info	11✔
238
239	@property	11✔
240	def known_scope_to_scoped_args(self) -> dict[str, frozenset[str]]:	11✔
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:	11✔
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):	11✔
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:	11✔
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:	11✔
300	"""Register an option in the given scope."""
301	self.get_registrar(scope).register(args, *kwargs)	11✔
302	deprecated_scope = self.known_scope_to_info[scope].deprecated_scope	11✔
303	if deprecated_scope:	11✔
UNCOV 304	self.get_registrar(deprecated_scope).register(args, *kwargs)	×
305
306	def get_registrar(self, scope: str) -> OptionRegistrar:	11✔
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	"""
313	try:	11✔
314	return self._registrar_by_scope[scope]	11✔
315	except KeyError:	×
316	raise ConfigValidationError(f"No such options scope: {scope}")	×
317
318	def _check_and_apply_deprecations(self, scope, values):	11✔
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	"""
330	si = self.known_scope_to_info[scope]	11✔
331
332	# If this Scope is itself deprecated, report that.
333	if si.removal_version:	11✔
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.
346	deprecated_scope = si.deprecated_scope	11✔
347	if deprecated_scope is not None and scope != deprecated_scope:	11✔
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	11✔
369	def for_scope(	11✔
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	"""
385	builder = OptionValueContainerBuilder()	11✔
386	mutex_map = defaultdict(list)	11✔
387	registrar = self.get_registrar(scope)	11✔
388	scope_str = "global scope" if scope == GLOBAL_SCOPE else f"scope '{scope}'"	11✔
389
390	for option_info in registrar.option_registrations_iter():	11✔
391	dest = option_info.kwargs["dest"]	11✔
392	val, rank = self._native_parser.get_value(scope=scope, option_info=option_info)	11✔
393	explicitly_set = rank > Rank.HARDCODED	11✔
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.
397	removal_version = option_info.kwargs.get("removal_version", None)	11✔
398	if removal_version is not None:	11✔
399	warn_or_error(	11✔
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.
408	if explicitly_set:	11✔
409	mutex_dest = option_info.kwargs.get("mutually_exclusive_group")	11✔
410	mutex_map_key = mutex_dest or dest	11✔
411	mutex_map[mutex_map_key].append(dest)	11✔
412	if len(mutex_map[mutex_map_key]) > 1:	11✔
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	)
422	setattr(builder, dest, RankedValue(rank, val))	11✔
423
424	# Check for any deprecation conditions, which are evaluated using `self._flag_matchers`.
425	if check_deprecations:	11✔
426	self._check_and_apply_deprecations(scope, builder)	11✔
427	return builder.build()	11✔
428
429	def get_fingerprintable_for_scope(	11✔
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
446	pairs = []	1✔
447	registrar = self.get_registrar(scope)	1✔
448	# Sort the arguments, so that the fingerprint is consistent.
449	for option_info in sorted(registrar.option_registrations_iter()):	1✔
450	if not option_info.kwargs.get("fingerprint", True):	1✔
451	continue	1✔
452	if daemon_only and not option_info.kwargs.get("daemon", False):	1✔
UNCOV 453	continue	×
454	dest = option_info.kwargs["dest"]	1✔
455	val = self.for_scope(scope)[dest]	1✔
456	# If we have a list then we delegate to the fingerprinting implementation of the members.
457	if is_list_option(option_info.kwargs):	1✔
458	val_type = option_info.kwargs.get("member_type", str)	1✔
459	else:
460	val_type = option_info.kwargs.get("type", str)	1✔
461	pairs.append((dest, val_type, val))	1✔
462	return pairs	1✔
463
464	def __getitem__(self, scope: str) -> OptionValueContainer:	11✔
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:	11✔
470	"""Return the option values for the global scope.
471
472	:API: public
473	"""
474	return self.for_scope(GLOBAL_SCOPE)	11✔

pantsbuild / pants / 19529437518

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