21803785359

Committed 08 Feb 2026 07:13PM UTC coverage: 43.3% (-37.0%) from 80.277%

Build # 21803785359

Build Type

Pull #23085

github

Committed by

web-flow

Commit Message

Merge 7c1cd926d into 40389cc58

Pull Request Pull Request #23085: A helper method for indexing paths by source root

Run Details

2 of 6 new or added lines in 1 file covered. (33.33%)

17114 existing lines in 539 files now uncovered.

26075 of 60219 relevant lines covered (43.3%)

0.43 hits per line

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

79.31

/src/python/pants/option/subsystem.py

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

from __future__ import annotations

import functools
import inspect
import re
from abc import ABCMeta
from collections.abc import Callable, Iterable, Sequence
from typing import TYPE_CHECKING, Any, ClassVar, TypeVar, cast

from pants.core.util_rules.env_vars import environment_vars_subset
from pants.engine.env_vars import EnvironmentVars, EnvironmentVarsRequest
from pants.engine.internals.options_parsing import scope_options
from pants.engine.internals.selectors import AwaitableConstraints
from pants.engine.rules import implicitly
from pants.engine.unions import UnionMembership, UnionRule, distinct_union_type_per_subclass
from pants.option.errors import OptionsError
from pants.option.option_types import OptionInfo, collect_options_info
from pants.option.option_value_container import OptionValueContainer
from pants.option.options import Options
from pants.option.scope import Scope, ScopedOptions, ScopeInfo, normalize_scope
from pants.util.frozendict import FrozenDict
from pants.util.strutil import softwrap

if TYPE_CHECKING:
    # Needed to avoid an import cycle.
    from pants.core.environments.target_types import EnvironmentTarget
    from pants.engine.rules import Rule

_SubsystemT = TypeVar("_SubsystemT", bound="Subsystem")


class _SubsystemMeta(ABCMeta):
    """Metaclass to link inner `EnvironmentAware` class with the enclosing subsystem."""

    def __init__(self, name, bases, namespace, **k):
        super().__init__(name, bases, namespace, **k)
        if (
            not (name == "Subsystem" and bases == ())
            and self.EnvironmentAware is not Subsystem.EnvironmentAware
        ):
            # Only `EnvironmentAware` subclasses should be linked to their enclosing scope
            if Subsystem.EnvironmentAware not in self.EnvironmentAware.__bases__:
                # Allow for `self.EnvironmentAware` to not need to explicitly derive from
                # `Subsystem.EnvironmentAware` (saving needless repetitive typing)
                self.EnvironmentAware = type(
                    "EnvironmentAware",
                    (
                        self.EnvironmentAware,
                        Subsystem.EnvironmentAware,
                        *self.EnvironmentAware.__bases__,
                    ),
                    {},
                )
            # A marker that allows initialization code to check for EnvironmentAware subclasses
            # without having to import Subsystem.EnvironmentAware and use issubclass(), which
            # can cause a dependency cycle.
            self.EnvironmentAware.__subsystem_environment_aware__ = True

            self.EnvironmentAware.subsystem = self


class Subsystem(metaclass=_SubsystemMeta):
    """A separable piece of functionality that may be reused across multiple tasks or other code.

    Subsystems encapsulate the configuration and initialization of things like JVMs,
    Python interpreters, SCMs and so on.

    Set the `help` class property with a description, which will be used in `./pants help`. For the
    best rendering, use soft wrapping (e.g. implicit string concatenation) within paragraphs, but
    hard wrapping (`\n`) to separate distinct paragraphs and/or lists.
    """

    # A marker that allows initialization code to check for Subsystem subclasses without
    # having to import Subsystem and use issubclass(), which can cause a dependency cycle.
    __subsystem__ = True

    options_scope: str
    help: ClassVar[str | Callable[[], str]]

    # Subclasses may override these to specify a deprecated former name for this Subsystem's scope.
    # Option values can be read from the deprecated scope, but a deprecation warning will be issued.
    # The deprecation warning becomes an error at the given Pants version (which must therefore be
    # a valid semver).
    deprecated_options_scope: str | None = None
    deprecated_options_scope_removal_version: str | None = None

    # // Note: must be aligned with the regex in src/rust/options/src/id.rs.
    _scope_name_re = re.compile(r"^(?:[a-z0-9_])+(?:-(?:[a-z0-9_])+)*$")

    _rules: ClassVar[Sequence[Rule] | None] = None

    class EnvironmentAware(metaclass=ABCMeta):
        """A separate container for options that may be redefined by the runtime environment.

        To define environment-aware options, create an inner class in the `Subsystem` called
        `EnvironmentAware`. Option fields share their scope with their enclosing `Subsystem`,
        and the values of fields will default to the values set through Pants' configuration.

        To consume environment-aware options, inject the `EnvironmentAware` inner class into
        your rule.

        Optionally, it is possible to specify environment variables that are required when
        post-processing raw values provided by users (e.g. `<PATH>` special strings) by specifying
        `env_vars_used_by_options`, and consuming `_options_env` in your post-processing property.
        These environment variables will be requested at construction time.
        """

        subsystem: ClassVar[type[Subsystem]]
        env_vars_used_by_options: ClassVar[tuple[str, ...]] = ()

        options: OptionValueContainer
        env_tgt: EnvironmentTarget
        _options_env: EnvironmentVars = EnvironmentVars()

        def __getattribute__(self, __name: str) -> Any:
            from pants.core.environments.rules import resolve_environment_sensitive_option

            # Will raise an `AttributeError` if the attribute is not defined.
            # MyPy should stop that from ever happening.
            default = super().__getattribute__(__name)

            # Check to see whether there's a definition of this attribute at the class level.
            # If it returns `default` then the attribute on the instance is the same object
            # as defined at the class, or the attribute does not exist on the class,
            # and we don't really need to go any further.
            v = getattr(type(self), __name, default)
            if v is default:
                return default

            # Resolving an attribute on the class object will return the underlying descriptor.
            # If the descriptor is an `OptionInfo`, we can resolve it against the environment
            # target.
            if isinstance(v, OptionInfo):
                # If the value is not defined in the `EnvironmentTarget`, return the value
                # from the options system.
                override = resolve_environment_sensitive_option(v.args[0], self)
                return override if override is not None else default

            # We should just return the default at this point.
            return default

        def _is_default(self, __name: str) -> bool:
            """Returns true if the value of the named option is unchanged from the default."""
            from pants.core.environments.rules import resolve_environment_sensitive_option

            v = getattr(type(self), __name)
            assert isinstance(v, OptionInfo)

            return (
                # vars beginning with `_` are exposed as option names with the leading `_` stripped
                self.options.is_default(__name.lstrip("_"))
                and resolve_environment_sensitive_option(v.args[0], self) is None
            )

    @classmethod
    def rules(cls: Any) -> Iterable[Rule]:
        # NB: This avoids using `memoized_classmethod` until its interaction with `mypy` can be improved.
        if cls._rules is None:
            from pants.core.environments.rules import add_option_fields_for
            from pants.engine.rules import Rule

            # nb. `rules` needs to be memoized so that repeated calls to add these rules
            # return exactly the same rule objects. As such, returning this generator
            # directly won't work, because the iterator needs to be replayable.
            def inner() -> Iterable[Rule]:
                yield cls._construct_subsystem_rule()
                if cls.EnvironmentAware is not Subsystem.EnvironmentAware:
                    yield cls._construct_env_aware_rule()
                    yield from (cast(Rule, i) for i in add_option_fields_for(cls.EnvironmentAware))

            cls._rules = tuple(inner())
        return cast("Sequence[Rule]", cls._rules)

    @distinct_union_type_per_subclass
    class PluginOption:
        pass

    @classmethod
    def register_plugin_options(cls, options_container: type) -> UnionRule:
        """Register additional options on the subsystem.

        In the `rules()` register.py entry-point, include `OtherSubsystem.register_plugin_options(<OptionsContainer>)`.
        `<OptionsContainer>` should be a type with option class attributes, similar to how they are
        defined for subsystems.

        This will register the option as a first-class citizen.
        Plugins can use this new option like any other.
        """
        return UnionRule(cls.PluginOption, options_container)

    @classmethod
    def _construct_subsystem_rule(cls) -> Rule:
        """Returns a `TaskRule` that will construct the target Subsystem."""

        # Global-level imports are conditional, we need to re-import here for runtime use
        from pants.engine.rules import TaskRule

        partial_construct_subsystem: Any = functools.partial(_construct_subsystem, cls)

        # NB: We must populate several dunder methods on the partial function because partial
        # functions do not have these defined by default and the engine uses these values to
        # visualize functions in error messages and the rule graph.
        snake_scope = normalize_scope(cls.options_scope)
        name = f"construct_scope_{snake_scope}"
        partial_construct_subsystem.__name__ = name
        partial_construct_subsystem.__module__ = cls.__module__
        partial_construct_subsystem.__doc__ = cls.help

        _, class_definition_lineno = inspect.getsourcelines(cls)
        partial_construct_subsystem.__line_number__ = class_definition_lineno

        return TaskRule(
            output_type=cls,
            parameters=FrozenDict(),
            awaitables=(
                AwaitableConstraints(
                    rule_id="pants.engine.internals.options_parsing.scope_options",
                    output_type=ScopedOptions,
                    explicit_args_arity=1,
                    # NB: For a call-by-name, the input_types are the explicit ones provided
                    # in a map passed to **implicitly(), and our call to this rule uses a
                    # positional arg and an empty **implicitly(), so input_types are empty here.
                    input_types=tuple(),
                ),
            ),
            masked_types=(),
            func=partial_construct_subsystem,
            canonical_name=name,
        )

    @classmethod
    def _construct_env_aware_rule(cls) -> Rule:
        """Returns a `TaskRule` that will construct the target Subsystem.EnvironmentAware."""
        # Global-level imports are conditional, we need to re-import here for runtime use
        from pants.core.environments.target_types import EnvironmentTarget
        from pants.engine.rules import TaskRule

        snake_scope = normalize_scope(cls.options_scope)
        name = f"construct_env_aware_scope_{snake_scope}"

        # placate the rule graph visualizer.
        @functools.wraps(_construct_env_aware)
        async def inner(*a, **k):
            return await _construct_env_aware(*a, **k)

        inner.__line_number__ = 0  # type: ignore[attr-defined]

        return TaskRule(
            output_type=cls.EnvironmentAware,
            parameters=FrozenDict({"subsystem_instance": cls, "env_tgt": EnvironmentTarget}),
            awaitables=(
                AwaitableConstraints(
                    rule_id="pants.core.util_rules.env_vars.environment_vars_subset",
                    output_type=EnvironmentVars,
                    explicit_args_arity=1,
                    # NB: For a call-by-name, the input_types are the explicit ones provided
                    # in a map passed to **implicitly(), and our call to this rule uses a
                    # positional arg and an empty **implicitly(), so input_types are empty here.
                    input_types=tuple(),
                ),
            ),
            masked_types=(),
            func=inner,
            canonical_name=name,
        )

    @classmethod
    def is_valid_scope_name(cls, s: str) -> bool:
        return s == "" or (cls._scope_name_re.match(s) is not None and s != "pants")

    @classmethod
    def validate_scope(cls) -> None:
        options_scope = getattr(cls, "options_scope", None)
        if options_scope is None:
            raise OptionsError(f"{cls.__name__} must set options_scope.")
        if not cls.is_valid_scope_name(options_scope):
            raise OptionsError(
                softwrap(
                    f"""
                    Options scope "{options_scope}" is not valid.

                    Replace in code with a new scope name consisting of only lower-case letters,
                    digits, underscores, and non-consecutive dashes.
                    """
                )
            )

    @classmethod
    def create_scope_info(cls, **scope_info_kwargs) -> ScopeInfo:
        """One place to create scope info, to allow subclasses to inject custom scope args."""
        return ScopeInfo(**scope_info_kwargs)

    @classmethod
    def get_scope_info(cls) -> ScopeInfo:
        """Returns a ScopeInfo instance representing this Subsystem's options scope."""
        cls.validate_scope()
        return cls.create_scope_info(scope=cls.options_scope, subsystem_cls=cls)

    @classmethod
    def register_options_on_scope(cls, options: Options, union_membership: UnionMembership):
        """Trigger registration of this Subsystem's options.

        Subclasses should not generally need to override this method.
        """

        def register(*args, **kwargs):
            options.register(cls.options_scope, *args, **kwargs)

        plugin_option_containers = union_membership.get(cls.PluginOption)
        for options_info in collect_options_info(cls):
            register(*options_info.args, **options_info.kwargs)
        for options_info in collect_options_info(cls.EnvironmentAware):
            register(*options_info.args, environment_aware=True, **options_info.kwargs)
        for options_info in (
            option
            for container in plugin_option_containers
            for option in collect_options_info(container)
        ):
            register(*options_info.args, **options_info.kwargs)

    def __init__(self, options: OptionValueContainer) -> None:
        self.validate_scope()
        self.options = options

    def __eq__(self, other: Any) -> bool:
        if type(self) != type(other):  # noqa: E721
            return False
        return bool(self.options == other.options)


async def _construct_subsystem(subsystem_typ: type[_SubsystemT]) -> _SubsystemT:
    scoped_options = await scope_options(Scope(str(subsystem_typ.options_scope)), **implicitly())
    return subsystem_typ(scoped_options.options)


async def _construct_env_aware(
    subsystem_instance: _SubsystemT,
    env_tgt: EnvironmentTarget,
) -> Subsystem.EnvironmentAware:
    t: Subsystem.EnvironmentAware = type(subsystem_instance).EnvironmentAware()
    # `_SubSystemMeta` metaclass should ensure that `EnvironmentAware` actually subclasses
    # `EnvironmentAware`, but if an implementer does something egregious, it's best we
    # catch it.
    assert isinstance(t, Subsystem.EnvironmentAware)

    t.options = subsystem_instance.options
    t.env_tgt = env_tgt

    if t.env_vars_used_by_options:
        t._options_env = await environment_vars_subset(
            EnvironmentVarsRequest(t.env_vars_used_by_options), **implicitly()
        )

    return t

1	# Copyright 2015 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 functools	1✔
7	import inspect	1✔
8	import re	1✔
9	from abc import ABCMeta	1✔
10	from collections.abc import Callable, Iterable, Sequence	1✔
11	from typing import TYPE_CHECKING, Any, ClassVar, TypeVar, cast	1✔
12
13	from pants.core.util_rules.env_vars import environment_vars_subset	1✔
14	from pants.engine.env_vars import EnvironmentVars, EnvironmentVarsRequest	1✔
15	from pants.engine.internals.options_parsing import scope_options	1✔
16	from pants.engine.internals.selectors import AwaitableConstraints	1✔
17	from pants.engine.rules import implicitly	1✔
18	from pants.engine.unions import UnionMembership, UnionRule, distinct_union_type_per_subclass	1✔
19	from pants.option.errors import OptionsError	1✔
20	from pants.option.option_types import OptionInfo, collect_options_info	1✔
21	from pants.option.option_value_container import OptionValueContainer	1✔
22	from pants.option.options import Options	1✔
23	from pants.option.scope import Scope, ScopedOptions, ScopeInfo, normalize_scope	1✔
24	from pants.util.frozendict import FrozenDict	1✔
25	from pants.util.strutil import softwrap	1✔
26
27	if TYPE_CHECKING:
28	# Needed to avoid an import cycle.
29	from pants.core.environments.target_types import EnvironmentTarget
30	from pants.engine.rules import Rule
31
32	_SubsystemT = TypeVar("_SubsystemT", bound="Subsystem")	1✔
33
34
35	class _SubsystemMeta(ABCMeta):	1✔
36	"""Metaclass to link inner `EnvironmentAware` class with the enclosing subsystem."""
37
38	def __init__(self, name, bases, namespace, **k):	1✔
39	super().__init__(name, bases, namespace, **k)	1✔
40	if (	1✔
41	not (name == "Subsystem" and bases == ())
42	and self.EnvironmentAware is not Subsystem.EnvironmentAware
43	):
44	# Only `EnvironmentAware` subclasses should be linked to their enclosing scope
45	if Subsystem.EnvironmentAware not in self.EnvironmentAware.__bases__:	1✔
46	# Allow for `self.EnvironmentAware` to not need to explicitly derive from
47	# `Subsystem.EnvironmentAware` (saving needless repetitive typing)
48	self.EnvironmentAware = type(	1✔
49	"EnvironmentAware",
50	(
51	self.EnvironmentAware,
52	Subsystem.EnvironmentAware,
53	*self.EnvironmentAware.__bases__,
54	),
55	{},
56	)
57	# A marker that allows initialization code to check for EnvironmentAware subclasses
58	# without having to import Subsystem.EnvironmentAware and use issubclass(), which
59	# can cause a dependency cycle.
60	self.EnvironmentAware.__subsystem_environment_aware__ = True	1✔
61
62	self.EnvironmentAware.subsystem = self	1✔
63
64
65	class Subsystem(metaclass=_SubsystemMeta):	1✔
66	"""A separable piece of functionality that may be reused across multiple tasks or other code.
67
68	Subsystems encapsulate the configuration and initialization of things like JVMs,
69	Python interpreters, SCMs and so on.
70
71	Set the `help` class property with a description, which will be used in `./pants help`. For the
72	best rendering, use soft wrapping (e.g. implicit string concatenation) within paragraphs, but
73	hard wrapping (`\n`) to separate distinct paragraphs and/or lists.
74	"""
75
76	# A marker that allows initialization code to check for Subsystem subclasses without
77	# having to import Subsystem and use issubclass(), which can cause a dependency cycle.
78	__subsystem__ = True	1✔
79
80	options_scope: str	1✔
81	help: ClassVar[str \| Callable[[], str]]	1✔
82
83	# Subclasses may override these to specify a deprecated former name for this Subsystem's scope.
84	# Option values can be read from the deprecated scope, but a deprecation warning will be issued.
85	# The deprecation warning becomes an error at the given Pants version (which must therefore be
86	# a valid semver).
87	deprecated_options_scope: str \| None = None	1✔
88	deprecated_options_scope_removal_version: str \| None = None	1✔
89
90	# // Note: must be aligned with the regex in src/rust/options/src/id.rs.
91	_scope_name_re = re.compile(r"^(?:[a-z0-9_])+(?:-(?:[a-z0-9_])+)*$")	1✔
92
93	_rules: ClassVar[Sequence[Rule] \| None] = None	1✔
94
95	class EnvironmentAware(metaclass=ABCMeta):	1✔
96	"""A separate container for options that may be redefined by the runtime environment.
97
98	To define environment-aware options, create an inner class in the `Subsystem` called
99	`EnvironmentAware`. Option fields share their scope with their enclosing `Subsystem`,
100	and the values of fields will default to the values set through Pants' configuration.
101
102	To consume environment-aware options, inject the `EnvironmentAware` inner class into
103	your rule.
104
105	Optionally, it is possible to specify environment variables that are required when
106	post-processing raw values provided by users (e.g. `<PATH>` special strings) by specifying
107	`env_vars_used_by_options`, and consuming `_options_env` in your post-processing property.
108	These environment variables will be requested at construction time.
109	"""
110
111	subsystem: ClassVar[type[Subsystem]]	1✔
112	env_vars_used_by_options: ClassVar[tuple[str, ...]] = ()	1✔
113
114	options: OptionValueContainer	1✔
115	env_tgt: EnvironmentTarget	1✔
116	_options_env: EnvironmentVars = EnvironmentVars()	1✔
117
118	def __getattribute__(self, __name: str) -> Any:	1✔
UNCOV 119	from pants.core.environments.rules import resolve_environment_sensitive_option	×
120
121	# Will raise an `AttributeError` if the attribute is not defined.
122	# MyPy should stop that from ever happening.
UNCOV 123	default = super().__getattribute__(__name)	×
124
125	# Check to see whether there's a definition of this attribute at the class level.
126	# If it returns `default` then the attribute on the instance is the same object
127	# as defined at the class, or the attribute does not exist on the class,
128	# and we don't really need to go any further.
UNCOV 129	v = getattr(type(self), __name, default)	×
UNCOV 130	if v is default:	×
UNCOV 131	return default	×
132
133	# Resolving an attribute on the class object will return the underlying descriptor.
134	# If the descriptor is an `OptionInfo`, we can resolve it against the environment
135	# target.
136	if isinstance(v, OptionInfo):	×
137	# If the value is not defined in the `EnvironmentTarget`, return the value
138	# from the options system.
139	override = resolve_environment_sensitive_option(v.args[0], self)	×
140	return override if override is not None else default	×
141
142	# We should just return the default at this point.
143	return default	×
144
145	def _is_default(self, __name: str) -> bool:	1✔
146	"""Returns true if the value of the named option is unchanged from the default."""
147	from pants.core.environments.rules import resolve_environment_sensitive_option	×
148
149	v = getattr(type(self), __name)	×
150	assert isinstance(v, OptionInfo)	×
151
152	return (	×
153	# vars beginning with `_` are exposed as option names with the leading `_` stripped
154	self.options.is_default(__name.lstrip("_"))
155	and resolve_environment_sensitive_option(v.args[0], self) is None
156	)
157
158	@classmethod	1✔
159	def rules(cls: Any) -> Iterable[Rule]:	1✔
160	# NB: This avoids using `memoized_classmethod` until its interaction with `mypy` can be improved.
161	if cls._rules is None:	1✔
162	from pants.core.environments.rules import add_option_fields_for	1✔
163	from pants.engine.rules import Rule	1✔
164
165	# nb. `rules` needs to be memoized so that repeated calls to add these rules
166	# return exactly the same rule objects. As such, returning this generator
167	# directly won't work, because the iterator needs to be replayable.
168	def inner() -> Iterable[Rule]:	1✔
169	yield cls._construct_subsystem_rule()	1✔
170	if cls.EnvironmentAware is not Subsystem.EnvironmentAware:	1✔
171	yield cls._construct_env_aware_rule()	1✔
172	yield from (cast(Rule, i) for i in add_option_fields_for(cls.EnvironmentAware))	1✔
173
174	cls._rules = tuple(inner())	1✔
175	return cast("Sequence[Rule]", cls._rules)	1✔
176
177	@distinct_union_type_per_subclass	1✔
178	class PluginOption:	1✔
179	pass	1✔
180
181	@classmethod	1✔
182	def register_plugin_options(cls, options_container: type) -> UnionRule:	1✔
183	"""Register additional options on the subsystem.
184
185	In the `rules()` register.py entry-point, include `OtherSubsystem.register_plugin_options(<OptionsContainer>)`.
186	`<OptionsContainer>` should be a type with option class attributes, similar to how they are
187	defined for subsystems.
188
189	This will register the option as a first-class citizen.
190	Plugins can use this new option like any other.
191	"""
UNCOV 192	return UnionRule(cls.PluginOption, options_container)	×
193
194	@classmethod	1✔
195	def _construct_subsystem_rule(cls) -> Rule:	1✔
196	"""Returns a `TaskRule` that will construct the target Subsystem."""
197
198	# Global-level imports are conditional, we need to re-import here for runtime use
199	from pants.engine.rules import TaskRule	1✔
200
201	partial_construct_subsystem: Any = functools.partial(_construct_subsystem, cls)	1✔
202
203	# NB: We must populate several dunder methods on the partial function because partial
204	# functions do not have these defined by default and the engine uses these values to
205	# visualize functions in error messages and the rule graph.
206	snake_scope = normalize_scope(cls.options_scope)	1✔
207	name = f"construct_scope_{snake_scope}"	1✔
208	partial_construct_subsystem.__name__ = name	1✔
209	partial_construct_subsystem.__module__ = cls.__module__	1✔
210	partial_construct_subsystem.__doc__ = cls.help	1✔
211
212	_, class_definition_lineno = inspect.getsourcelines(cls)	1✔
213	partial_construct_subsystem.__line_number__ = class_definition_lineno	1✔
214
215	return TaskRule(	1✔
216	output_type=cls,
217	parameters=FrozenDict(),
218	awaitables=(
219	AwaitableConstraints(
220	rule_id="pants.engine.internals.options_parsing.scope_options",
221	output_type=ScopedOptions,
222	explicit_args_arity=1,
223	# NB: For a call-by-name, the input_types are the explicit ones provided
224	# in a map passed to **implicitly(), and our call to this rule uses a
225	# positional arg and an empty **implicitly(), so input_types are empty here.
226	input_types=tuple(),
227	),
228	),
229	masked_types=(),
230	func=partial_construct_subsystem,
231	canonical_name=name,
232	)
233
234	@classmethod	1✔
235	def _construct_env_aware_rule(cls) -> Rule:	1✔
236	"""Returns a `TaskRule` that will construct the target Subsystem.EnvironmentAware."""
237	# Global-level imports are conditional, we need to re-import here for runtime use
238	from pants.core.environments.target_types import EnvironmentTarget	1✔
239	from pants.engine.rules import TaskRule	1✔
240
241	snake_scope = normalize_scope(cls.options_scope)	1✔
242	name = f"construct_env_aware_scope_{snake_scope}"	1✔
243
244	# placate the rule graph visualizer.
245	@functools.wraps(_construct_env_aware)	1✔
246	async def inner(a, *k):	1✔
247	return await _construct_env_aware(a, *k)	×
248
249	inner.__line_number__ = 0 # type: ignore[attr-defined]	1✔
250
251	return TaskRule(	1✔
252	output_type=cls.EnvironmentAware,
253	parameters=FrozenDict({"subsystem_instance": cls, "env_tgt": EnvironmentTarget}),
254	awaitables=(
255	AwaitableConstraints(
256	rule_id="pants.core.util_rules.env_vars.environment_vars_subset",
257	output_type=EnvironmentVars,
258	explicit_args_arity=1,
259	# NB: For a call-by-name, the input_types are the explicit ones provided
260	# in a map passed to **implicitly(), and our call to this rule uses a
261	# positional arg and an empty **implicitly(), so input_types are empty here.
262	input_types=tuple(),
263	),
264	),
265	masked_types=(),
266	func=inner,
267	canonical_name=name,
268	)
269
270	@classmethod	1✔
271	def is_valid_scope_name(cls, s: str) -> bool:	1✔
272	return s == "" or (cls._scope_name_re.match(s) is not None and s != "pants")	1✔
273
274	@classmethod	1✔
275	def validate_scope(cls) -> None:	1✔
276	options_scope = getattr(cls, "options_scope", None)	1✔
277	if options_scope is None:	1✔
UNCOV 278	raise OptionsError(f"{cls.__name__} must set options_scope.")	×
279	if not cls.is_valid_scope_name(options_scope):	1✔
280	raise OptionsError(	×
281	softwrap(
282	f"""
283	Options scope "{options_scope}" is not valid.
284
285	Replace in code with a new scope name consisting of only lower-case letters,
286	digits, underscores, and non-consecutive dashes.
287	"""
288	)
289	)
290
291	@classmethod	1✔
292	def create_scope_info(cls, **scope_info_kwargs) -> ScopeInfo:	1✔
293	"""One place to create scope info, to allow subclasses to inject custom scope args."""
294	return ScopeInfo(**scope_info_kwargs)	1✔
295
296	@classmethod	1✔
297	def get_scope_info(cls) -> ScopeInfo:	1✔
298	"""Returns a ScopeInfo instance representing this Subsystem's options scope."""
299	cls.validate_scope()	1✔
300	return cls.create_scope_info(scope=cls.options_scope, subsystem_cls=cls)	1✔
301
302	@classmethod	1✔
303	def register_options_on_scope(cls, options: Options, union_membership: UnionMembership):	1✔
304	"""Trigger registration of this Subsystem's options.
305
306	Subclasses should not generally need to override this method.
307	"""
308
309	def register(args, *kwargs):	1✔
310	options.register(cls.options_scope, args, *kwargs)	1✔
311
312	plugin_option_containers = union_membership.get(cls.PluginOption)	1✔
313	for options_info in collect_options_info(cls):	1✔
314	register(options_info.args, *options_info.kwargs)	1✔
315	for options_info in collect_options_info(cls.EnvironmentAware):	1✔
316	register(options_info.args, environment_aware=True, *options_info.kwargs)	1✔
317	for options_info in (	1✔
318	option
319	for container in plugin_option_containers
320	for option in collect_options_info(container)
321	):
UNCOV 322	register(options_info.args, *options_info.kwargs)	×
323
324	def __init__(self, options: OptionValueContainer) -> None:	1✔
325	self.validate_scope()	1✔
326	self.options = options	1✔
327
328	def __eq__(self, other: Any) -> bool:	1✔
329	if type(self) != type(other): # noqa: E721	×
330	return False	×
331	return bool(self.options == other.options)	×
332
333
334	async def _construct_subsystem(subsystem_typ: type[_SubsystemT]) -> _SubsystemT:	1✔
UNCOV 335	scoped_options = await scope_options(Scope(str(subsystem_typ.options_scope)), **implicitly())	×
UNCOV 336	return subsystem_typ(scoped_options.options)	×
337
338
339	async def _construct_env_aware(	1✔
340	subsystem_instance: _SubsystemT,
341	env_tgt: EnvironmentTarget,
342	) -> Subsystem.EnvironmentAware:
343	t: Subsystem.EnvironmentAware = type(subsystem_instance).EnvironmentAware()	×
344	# `_SubSystemMeta` metaclass should ensure that `EnvironmentAware` actually subclasses
345	# `EnvironmentAware`, but if an implementer does something egregious, it's best we
346	# catch it.
347	assert isinstance(t, Subsystem.EnvironmentAware)	×
348
349	t.options = subsystem_instance.options	×
350	t.env_tgt = env_tgt	×
351
352	if t.env_vars_used_by_options:	×
353	t._options_env = await environment_vars_subset(	×
354	EnvironmentVarsRequest(t.env_vars_used_by_options), **implicitly()
355	)
356
357	return t	×

pantsbuild / pants / 21803785359

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