20147226056

Committed 11 Dec 2025 08:58PM UTC coverage: 78.827% (-1.5%) from 80.293%

Build # 20147226056

Build Type

push

github

Committed by

web-flow

Commit Message

Forwarded the `style` and `complete-platform` args from pants.toml to PEX (#22910)

## Context

After Apple switched to the `arm64` architecture, some package
publishers stopped releasing `x86_64` variants of their packages for
`darwin`. As a result, generating a universal lockfile now fails because
no single package version is compatible with both `x86_64` and `arm64`
on `darwin`.

The solution is to use the `--style` and `--complete-platform` flags
with PEX. For example:
```
pex3 lock create \
    --style strict \
    --complete-platform 3rdparty/platforms/manylinux_2_28_aarch64.json \
    --complete-platform 3rdparty/platforms/macosx_26_0_arm64.json \
    -r 3rdparty/python/requirements_pyarrow.txt \
    -o python-pyarrow.lock
```

See the Slack discussion here:
https://pantsbuild.slack.com/archives/C046T6T9U/p1760098582461759

## Reproduction

* `BUILD`
```
python_requirement(
    name="awswrangler",
    requirements=["awswrangler==3.12.1"],
    resolve="awswrangler",
)
```
* Run `pants generate-lockfiles --resolve=awswrangler` on macOS with an
`arm64` CPU
```
pip: ERROR: Cannot install awswrangler==3.12.1 because these package versions have conflicting dependencies.
pip: ERROR: ResolutionImpossible: for help visit https://pip.pypa.io/en/latest/topics/dependency-resolution/#dealing-with-dependency-conflicts
pip:  
pip:  The conflict is caused by:
pip:      awswrangler 3.12.1 depends on pyarrow<18.0.0 and >=8.0.0; sys_platform == "darwin" and platform_machine == "x86_64"
pip:      awswrangler 3.12.1 depends on pyarrow<21.0.0 and >=18.0.0; sys_platform != "darwin" or platform_machine != "x86_64"
pip:  
pip:  Additionally, some packages in these conflicts have no matching distributions available for your environment:
pip:      pyarrow
pip:  
pip:  To fix this you could try to:
pip:  1. loosen the range of package versions you've specified
pip:  2. remove package versions to allow pip to attempt to solve the dependency conflict
```

## Implementation
... (continued)

Run Details

77 of 100 new or added lines in 6 files covered. (77.0%)

868 existing lines in 42 files now uncovered.

74471 of 94474 relevant lines covered (78.83%)

3.18 hits per line

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

84.83

/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	11✔
5
6	import functools	11✔
7	import inspect	11✔
8	import re	11✔
9	from abc import ABCMeta	11✔
10	from collections.abc import Callable, Iterable, Sequence	11✔
11	from typing import TYPE_CHECKING, Any, ClassVar, TypeVar, cast	11✔
12
13	from pants.core.util_rules.env_vars import environment_vars_subset	11✔
14	from pants.engine.env_vars import EnvironmentVars, EnvironmentVarsRequest	11✔
15	from pants.engine.internals.options_parsing import scope_options	11✔
16	from pants.engine.internals.selectors import AwaitableConstraints	11✔
17	from pants.engine.rules import implicitly	11✔
18	from pants.engine.unions import UnionMembership, UnionRule, distinct_union_type_per_subclass	11✔
19	from pants.option.errors import OptionsError	11✔
20	from pants.option.option_types import OptionInfo, collect_options_info	11✔
21	from pants.option.option_value_container import OptionValueContainer	11✔
22	from pants.option.options import Options	11✔
23	from pants.option.scope import Scope, ScopedOptions, ScopeInfo, normalize_scope	11✔
24	from pants.util.frozendict import FrozenDict	11✔
25	from pants.util.strutil import softwrap	11✔
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")	11✔
33
34
35	class _SubsystemMeta(ABCMeta):	11✔
36	"""Metaclass to link inner `EnvironmentAware` class with the enclosing subsystem."""
37
38	def __init__(self, name, bases, namespace, **k):	11✔
39	super().__init__(name, bases, namespace, **k)	11✔
40	if (	11✔
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__:	11✔
46	# Allow for `self.EnvironmentAware` to not need to explicitly derive from
47	# `Subsystem.EnvironmentAware` (saving needless repetitive typing)
48	self.EnvironmentAware = type(	11✔
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	11✔
61
62	self.EnvironmentAware.subsystem = self	11✔
63
64
65	class Subsystem(metaclass=_SubsystemMeta):	11✔
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	11✔
79
80	options_scope: str	11✔
81	help: ClassVar[str \| Callable[[], str]]	11✔
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	11✔
88	deprecated_options_scope_removal_version: str \| None = None	11✔
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_])+)*$")	11✔
92
93	_rules: ClassVar[Sequence[Rule] \| None] = None	11✔
94
95	class EnvironmentAware(metaclass=ABCMeta):	11✔
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]]	11✔
112	env_vars_used_by_options: ClassVar[tuple[str, ...]] = ()	11✔
113
114	options: OptionValueContainer	11✔
115	env_tgt: EnvironmentTarget	11✔
116	_options_env: EnvironmentVars = EnvironmentVars()	11✔
117
118	def __getattribute__(self, __name: str) -> Any:	11✔
119	from pants.core.environments.rules import resolve_environment_sensitive_option	1✔
120
121	# Will raise an `AttributeError` if the attribute is not defined.
122	# MyPy should stop that from ever happening.
123	default = super().__getattribute__(__name)	1✔
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.
129	v = getattr(type(self), __name, default)	1✔
130	if v is default:	1✔
131	return default	1✔
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:	11✔
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	11✔
159	def rules(cls: Any) -> Iterable[Rule]:	11✔
160	# NB: This avoids using `memoized_classmethod` until its interaction with `mypy` can be improved.
161	if cls._rules is None:	11✔
162	from pants.core.environments.rules import add_option_fields_for	11✔
163	from pants.engine.rules import Rule	11✔
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]:	11✔
169	yield cls._construct_subsystem_rule()	11✔
170	if cls.EnvironmentAware is not Subsystem.EnvironmentAware:	11✔
171	yield cls._construct_env_aware_rule()	11✔
172	yield from (cast(Rule, i) for i in add_option_fields_for(cls.EnvironmentAware))	11✔
173
174	cls._rules = tuple(inner())	11✔
175	return cast("Sequence[Rule]", cls._rules)	11✔
176
177	@distinct_union_type_per_subclass	11✔
178	class PluginOption:	11✔
179	pass	11✔
180
181	@classmethod	11✔
182	def register_plugin_options(cls, options_container: type) -> UnionRule:	11✔
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	"""
192	return UnionRule(cls.PluginOption, options_container)	5✔
193
194	@classmethod	11✔
195	def _construct_subsystem_rule(cls) -> Rule:	11✔
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	11✔
200
201	partial_construct_subsystem: Any = functools.partial(_construct_subsystem, cls)	11✔
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)	11✔
207	name = f"construct_scope_{snake_scope}"	11✔
208	partial_construct_subsystem.__name__ = name	11✔
209	partial_construct_subsystem.__module__ = cls.__module__	11✔
210	partial_construct_subsystem.__doc__ = cls.help	11✔
211
212	_, class_definition_lineno = inspect.getsourcelines(cls)	11✔
213	partial_construct_subsystem.__line_number__ = class_definition_lineno	11✔
214
215	return TaskRule(	11✔
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	11✔
236	def _construct_env_aware_rule(cls) -> Rule:	11✔
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
239	from pants.core.environments.target_types import EnvironmentTarget	11✔
240	from pants.engine.rules import TaskRule	11✔
241
242	snake_scope = normalize_scope(cls.options_scope)	11✔
243	name = f"construct_env_aware_scope_{snake_scope}"	11✔
244
245	# placate the rule graph visualizer.
246	@functools.wraps(_construct_env_aware)	11✔
247	async def inner(a, *k):	11✔
248	return await _construct_env_aware(a, *k)	×
249
250	inner.__line_number__ = 0 # type: ignore[attr-defined]	11✔
251
252	return TaskRule(	11✔
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	11✔
273	def is_valid_scope_name(cls, s: str) -> bool:	11✔
274	return s == "" or (cls._scope_name_re.match(s) is not None and s != "pants")	11✔
275
276	@classmethod	11✔
277	def validate_scope(cls) -> None:	11✔
278	options_scope = getattr(cls, "options_scope", None)	11✔
279	if options_scope is None:	11✔
280	raise OptionsError(f"{cls.__name__} must set options_scope.")	1✔
281	if not cls.is_valid_scope_name(options_scope):	11✔
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	11✔
294	def create_scope_info(cls, **scope_info_kwargs) -> ScopeInfo:	11✔
295	"""One place to create scope info, to allow subclasses to inject custom scope args."""
296	return ScopeInfo(**scope_info_kwargs)	11✔
297
298	@classmethod	11✔
299	def get_scope_info(cls) -> ScopeInfo:	11✔
300	"""Returns a ScopeInfo instance representing this Subsystem's options scope."""
301	cls.validate_scope()	11✔
302	return cls.create_scope_info(scope=cls.options_scope, subsystem_cls=cls)	11✔
303
304	@classmethod	11✔
305	def register_options_on_scope(cls, options: Options, union_membership: UnionMembership):	11✔
306	"""Trigger registration of this Subsystem's options.
307
308	Subclasses should not generally need to override this method.
309	"""
310
311	def register(args, *kwargs):	11✔
312	options.register(cls.options_scope, args, *kwargs)	11✔
313
314	plugin_option_containers = union_membership.get(cls.PluginOption)	11✔
315	for options_info in collect_options_info(cls):	11✔
316	register(options_info.args, *options_info.kwargs)	11✔
317	for options_info in collect_options_info(cls.EnvironmentAware):	11✔
318	register(options_info.args, environment_aware=True, *options_info.kwargs)	11✔
319	for options_info in (	11✔
320	option
321	for container in plugin_option_containers
322	for option in collect_options_info(container)
323	):
324	register(options_info.args, *options_info.kwargs)	3✔
325
326	def __init__(self, options: OptionValueContainer) -> None:	11✔
327	self.validate_scope()	11✔
328	self.options = options	11✔
329
330	def __eq__(self, other: Any) -> bool:	11✔
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:	11✔
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(	11✔
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 / 20147226056

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