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

48.28

/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(),
                    is_effect=False,
                ),
            ),
            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(),
                    is_effect=False,
                ),
            ),
            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.
UNCOV 161	if cls._rules is None:	×
UNCOV 162	from pants.core.environments.rules import add_option_fields_for	×
UNCOV 163	from pants.engine.rules import Rule	×
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.
UNCOV 168	def inner() -> Iterable[Rule]:	×
UNCOV 169	yield cls._construct_subsystem_rule()	×
UNCOV 170	if cls.EnvironmentAware is not Subsystem.EnvironmentAware:	×
UNCOV 171	yield cls._construct_env_aware_rule()	×
UNCOV 172	yield from (cast(Rule, i) for i in add_option_fields_for(cls.EnvironmentAware))	×
173
UNCOV 174	cls._rules = tuple(inner())	×
UNCOV 175	return cast("Sequence[Rule]", cls._rules)	×
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
UNCOV 199	from pants.engine.rules import TaskRule	×
200
UNCOV 201	partial_construct_subsystem: Any = functools.partial(_construct_subsystem, cls)	×
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.
UNCOV 206	snake_scope = normalize_scope(cls.options_scope)	×
UNCOV 207	name = f"construct_scope_{snake_scope}"	×
UNCOV 208	partial_construct_subsystem.__name__ = name	×
UNCOV 209	partial_construct_subsystem.__module__ = cls.__module__	×
UNCOV 210	partial_construct_subsystem.__doc__ = cls.help	×
211
UNCOV 212	_, class_definition_lineno = inspect.getsourcelines(cls)	×
UNCOV 213	partial_construct_subsystem.__line_number__ = class_definition_lineno	×
214
UNCOV 215	return TaskRule(	×
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	is_effect=False,
228	),
229	),
230	masked_types=(),
231	func=partial_construct_subsystem,
232	canonical_name=name,
233	)
234
235	@classmethod	1✔
236	def _construct_env_aware_rule(cls) -> Rule:	1✔
237	"""Returns a `TaskRule` that will construct the target Subsystem.EnvironmentAware."""
238	# Global-level imports are conditional, we need to re-import here for runtime use
UNCOV 239	from pants.core.environments.target_types import EnvironmentTarget	×
UNCOV 240	from pants.engine.rules import TaskRule	×
241
UNCOV 242	snake_scope = normalize_scope(cls.options_scope)	×
UNCOV 243	name = f"construct_env_aware_scope_{snake_scope}"	×
244
245	# placate the rule graph visualizer.
UNCOV 246	@functools.wraps(_construct_env_aware)	×
UNCOV 247	async def inner(a, *k):	×
248	return await _construct_env_aware(a, *k)	×
249
UNCOV 250	inner.__line_number__ = 0 # type: ignore[attr-defined]	×
251
UNCOV 252	return TaskRule(	×
253	output_type=cls.EnvironmentAware,
254	parameters=FrozenDict({"subsystem_instance": cls, "env_tgt": EnvironmentTarget}),
255	awaitables=(
256	AwaitableConstraints(
257	rule_id="pants.core.util_rules.env_vars.environment_vars_subset",
258	output_type=EnvironmentVars,
259	explicit_args_arity=1,
260	# NB: For a call-by-name, the input_types are the explicit ones provided
261	# in a map passed to **implicitly(), and our call to this rule uses a
262	# positional arg and an empty **implicitly(), so input_types are empty here.
263	input_types=tuple(),
264	is_effect=False,
265	),
266	),
267	masked_types=(),
268	func=inner,
269	canonical_name=name,
270	)
271
272	@classmethod	1✔
273	def is_valid_scope_name(cls, s: str) -> bool:	1✔
UNCOV 274	return s == "" or (cls._scope_name_re.match(s) is not None and s != "pants")	×
275
276	@classmethod	1✔
277	def validate_scope(cls) -> None:	1✔
UNCOV 278	options_scope = getattr(cls, "options_scope", None)	×
UNCOV 279	if options_scope is None:	×
UNCOV 280	raise OptionsError(f"{cls.__name__} must set options_scope.")	×
UNCOV 281	if not cls.is_valid_scope_name(options_scope):	×
282	raise OptionsError(	×
283	softwrap(
284	f"""
285	Options scope "{options_scope}" is not valid.
286
287	Replace in code with a new scope name consisting of only lower-case letters,
288	digits, underscores, and non-consecutive dashes.
289	"""
290	)
291	)
292
293	@classmethod	1✔
294	def create_scope_info(cls, **scope_info_kwargs) -> ScopeInfo:	1✔
295	"""One place to create scope info, to allow subclasses to inject custom scope args."""
UNCOV 296	return ScopeInfo(**scope_info_kwargs)	×
297
298	@classmethod	1✔
299	def get_scope_info(cls) -> ScopeInfo:	1✔
300	"""Returns a ScopeInfo instance representing this Subsystem's options scope."""
UNCOV 301	cls.validate_scope()	×
UNCOV 302	return cls.create_scope_info(scope=cls.options_scope, subsystem_cls=cls)	×
303
304	@classmethod	1✔
305	def register_options_on_scope(cls, options: Options, union_membership: UnionMembership):	1✔
306	"""Trigger registration of this Subsystem's options.
307
308	Subclasses should not generally need to override this method.
309	"""
310
UNCOV 311	def register(args, *kwargs):	×
UNCOV 312	options.register(cls.options_scope, args, *kwargs)	×
313
UNCOV 314	plugin_option_containers = union_membership.get(cls.PluginOption)	×
UNCOV 315	for options_info in collect_options_info(cls):	×
UNCOV 316	register(options_info.args, *options_info.kwargs)	×
UNCOV 317	for options_info in collect_options_info(cls.EnvironmentAware):	×
UNCOV 318	register(options_info.args, environment_aware=True, *options_info.kwargs)	×
UNCOV 319	for options_info in (	×
320	option
321	for container in plugin_option_containers
322	for option in collect_options_info(container)
323	):
UNCOV 324	register(options_info.args, *options_info.kwargs)	×
325
326	def __init__(self, options: OptionValueContainer) -> None:	1✔
UNCOV 327	self.validate_scope()	×
UNCOV 328	self.options = options	×
329
330	def __eq__(self, other: Any) -> bool:	1✔
331	if type(self) != type(other): # noqa: E721	×
332	return False	×
333	return bool(self.options == other.options)	×
334
335
336	async def _construct_subsystem(subsystem_typ: type[_SubsystemT]) -> _SubsystemT:	1✔
UNCOV 337	scoped_options = await scope_options(Scope(str(subsystem_typ.options_scope)), **implicitly())	×
UNCOV 338	return subsystem_typ(scoped_options.options)	×
339
340
341	async def _construct_env_aware(	1✔
342	subsystem_instance: _SubsystemT,
343	env_tgt: EnvironmentTarget,
344	) -> Subsystem.EnvironmentAware:
345	t: Subsystem.EnvironmentAware = type(subsystem_instance).EnvironmentAware()	×
346	# `_SubSystemMeta` metaclass should ensure that `EnvironmentAware` actually subclasses
347	# `EnvironmentAware`, but if an implementer does something egregious, it's best we
348	# catch it.
349	assert isinstance(t, Subsystem.EnvironmentAware)	×
350
351	t.options = subsystem_instance.options	×
352	t.env_tgt = env_tgt	×
353
354	if t.env_vars_used_by_options:	×
355	t._options_env = await environment_vars_subset(	×
356	EnvironmentVarsRequest(t.env_vars_used_by_options), **implicitly()
357	)
358
359	return t	×

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