26080722777

Committed 19 May 2026 06:37AM UTC coverage: 52.106% (-11.5%) from 63.597%

Build # 26080722777

Build Type

Pull #23250

github

Committed by

web-flow

Commit Message

Merge 63ec06323 into 2693df832

Pull Request Pull Request #23250: Feature: Add generic option to docker image

Coverage Stats

12 of 50 new or added lines in 3 files covered. (24.0%)

5382 existing lines in 201 files now uncovered.

32053 of 61515 relevant lines covered (52.11%)

1.04 hits per line

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

43.42

/src/python/pants/init/options_initializer.py

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

from __future__ import annotations

import dataclasses
import importlib.util
import logging
import sys
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path

from pants.build_graph.build_configuration import BuildConfiguration
from pants.engine.env_vars import CompleteEnvironmentVars
from pants.engine.internals.native_engine import PyExecutor
from pants.engine.unions import UnionMembership
from pants.help.flag_error_help_printer import FlagErrorHelpPrinter
from pants.init.bootstrap_scheduler import BootstrapScheduler
from pants.init.engine_initializer import EngineInitializer
from pants.init.extension_loader import (
    load_backends_and_plugins,
    load_build_configuration_from_source,
)
from pants.init.plugin_resolver import PluginResolver
from pants.init.plugin_resolver import rules as plugin_resolver_rules
from pants.option.bootstrap_options import DynamicRemoteOptions
from pants.option.errors import UnknownFlagsError
from pants.option.options import Options
from pants.option.options_bootstrapper import OptionsBootstrapper
from pants.util.requirements import parse_requirements_file

logger = logging.getLogger(__name__)


def _initialize_build_configuration(
    plugin_resolver: PluginResolver,
    options_bootstrapper: OptionsBootstrapper,
    env: CompleteEnvironmentVars,
) -> BuildConfiguration:
    """Initialize a BuildConfiguration for the given OptionsBootstrapper.

    NB: This method:
      1. has the side effect of (idempotently) adding PYTHONPATH entries for this process
      2. is expensive to call, because it might resolve plugins from the network
    """

    bootstrap_options = options_bootstrapper.bootstrap_options.for_global_scope()

    # Add any extra paths to python path (e.g., for loading extra source backends).
    for path in bootstrap_options.pythonpath:
        if path not in sys.path:
            sys.path.append(path)

    # Resolve the actual Python code for any plugins. `sys.path` is modified as a side effect if
    # plugins were configured.
    backends_requirements = _collect_backends_requirements(bootstrap_options.backend_packages)
    plugin_resolver.resolve(options_bootstrapper, env, backends_requirements)

    # Load plugins and backends.
    return load_backends_and_plugins(
        bootstrap_options.plugins,
        bootstrap_options.backend_packages,
        BuildConfiguration.Builder(_pants_ng=bootstrap_options.pants_ng),
    )


def _collect_backends_requirements(backends: list[str]) -> list[str]:
    """Collects backend package dependencies, in case those are declared in an adjacent
    requirements.txt. Ignores any loading errors, assuming those will be later on handled by the
    backends loader.

    :param backends: An list of packages to load v2 backends requirements from.
    """
    requirements = []

    for backend_package in backends:
        try:
            backend_package_spec = importlib.util.find_spec(backend_package)
        except ModuleNotFoundError:
            continue

        if backend_package_spec is None:
            continue

        if backend_package_spec.origin is None:
            logger.warning(
                f"Can not check requirements for backend: '{backend_package}'. A __init__.py file is probably missing."
            )
            continue

        requirements_txt_file_path = Path(backend_package_spec.origin).parent.joinpath(
            "requirements.txt"
        )
        if requirements_txt_file_path.exists():
            content = requirements_txt_file_path.read_text()
            backend_package_requirements = [
                str(r)
                for r in parse_requirements_file(content, rel_path=str(requirements_txt_file_path))
            ]
            requirements.extend(backend_package_requirements)

    return requirements


def create_bootstrap_scheduler(
    options_bootstrapper: OptionsBootstrapper, executor: PyExecutor
) -> BootstrapScheduler:
    bc_builder = BuildConfiguration.Builder()
    # To load plugins, we only need access to the Python/PEX rules.
    load_build_configuration_from_source(bc_builder, ["pants.backend.python"])
    # And to plugin-loading-specific rules.
    bc_builder.register_rules("_dummy_for_bootstrapping_", plugin_resolver_rules())
    # We allow unrecognized options to defer any option error handling until post-bootstrap.
    bc_builder.allow_unknown_options()
    return BootstrapScheduler(
        EngineInitializer.setup_graph(
            options_bootstrapper.bootstrap_options.for_global_scope(),
            bc_builder.create(),
            DynamicRemoteOptions.disabled(),
            executor,
            is_bootstrap=True,
        ).scheduler
    )


class OptionsInitializer:
    """Initializes BuildConfiguration and Options instances given an OptionsBootstrapper.

    NB: Although this constructor takes an instance of the OptionsBootstrapper, it is
    used only to construct a "bootstrap" Scheduler: actual calls to resolve plugins use a
    per-request instance of the OptionsBootstrapper, which might request different plugins.

    TODO: We would eventually like to use the bootstrap Scheduler to construct the
    OptionsBootstrapper as well, but for now we do the opposite thing, and the Scheduler is
    used only to resolve plugins.
      see https://github.com/pantsbuild/pants/pull/11568
    """

    def __init__(
        self,
        options_bootstrapper: OptionsBootstrapper,
        executor: PyExecutor,
    ) -> None:
        self._bootstrap_scheduler = create_bootstrap_scheduler(options_bootstrapper, executor)
        self._plugin_resolver = PluginResolver(self._bootstrap_scheduler)

    def build_config(
        self,
        options_bootstrapper: OptionsBootstrapper,
        env: CompleteEnvironmentVars,
    ) -> BuildConfiguration:
        return _initialize_build_configuration(self._plugin_resolver, options_bootstrapper, env)

    def options(
        self,
        options_bootstrapper: OptionsBootstrapper,
        env: CompleteEnvironmentVars,
        build_config: BuildConfiguration,
        union_membership: UnionMembership,
        *,
        raise_: bool,
    ) -> Options:
        with self.handle_unknown_flags(options_bootstrapper, env, raise_=raise_):
            return options_bootstrapper.full_options(
                build_config.known_scope_infos, union_membership, build_config.allow_unknown_options
            )

    @contextmanager
    def handle_unknown_flags(
        self,
        options_bootstrapper: OptionsBootstrapper,
        env: CompleteEnvironmentVars,
        *,
        raise_: bool,
    ) -> Iterator[None]:
        """If there are any unknown flags, print "Did you mean?" and possibly error."""
        try:
            yield
        except UnknownFlagsError as err:
            build_config = _initialize_build_configuration(
                self._plugin_resolver, options_bootstrapper, env
            )
            # We need an options instance in order to get "did you mean" suggestions, but we know
            # there are bad flags in the args, so we generate options with no flags.
            no_arg_bootstrapper = dataclasses.replace(
                options_bootstrapper, args=("dummy_first_arg",)
            )
            options = no_arg_bootstrapper.full_options(
                build_config.known_scope_infos,
                union_membership=UnionMembership.empty(),
            )
            FlagErrorHelpPrinter(options).handle_unknown_flags(err)
            if raise_:
                raise err

1	# Copyright 2016 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	2✔
5
6	import dataclasses	2✔
7	import importlib.util	2✔
8	import logging	2✔
9	import sys	2✔
10	from collections.abc import Iterator	2✔
11	from contextlib import contextmanager	2✔
12	from pathlib import Path	2✔
13
14	from pants.build_graph.build_configuration import BuildConfiguration	2✔
15	from pants.engine.env_vars import CompleteEnvironmentVars	2✔
16	from pants.engine.internals.native_engine import PyExecutor	2✔
17	from pants.engine.unions import UnionMembership	2✔
18	from pants.help.flag_error_help_printer import FlagErrorHelpPrinter	2✔
19	from pants.init.bootstrap_scheduler import BootstrapScheduler	2✔
20	from pants.init.engine_initializer import EngineInitializer	2✔
21	from pants.init.extension_loader import (	2✔
22	load_backends_and_plugins,
23	load_build_configuration_from_source,
24	)
25	from pants.init.plugin_resolver import PluginResolver	2✔
26	from pants.init.plugin_resolver import rules as plugin_resolver_rules	2✔
27	from pants.option.bootstrap_options import DynamicRemoteOptions	2✔
28	from pants.option.errors import UnknownFlagsError	2✔
29	from pants.option.options import Options	2✔
30	from pants.option.options_bootstrapper import OptionsBootstrapper	2✔
31	from pants.util.requirements import parse_requirements_file	2✔
32
33	logger = logging.getLogger(__name__)	2✔
34
35
36	def _initialize_build_configuration(	2✔
37	plugin_resolver: PluginResolver,
38	options_bootstrapper: OptionsBootstrapper,
39	env: CompleteEnvironmentVars,
40	) -> BuildConfiguration:
41	"""Initialize a BuildConfiguration for the given OptionsBootstrapper.
42
43	NB: This method:
44	1. has the side effect of (idempotently) adding PYTHONPATH entries for this process
45	2. is expensive to call, because it might resolve plugins from the network
46	"""
47
UNCOV 48	bootstrap_options = options_bootstrapper.bootstrap_options.for_global_scope()	×
49
50	# Add any extra paths to python path (e.g., for loading extra source backends).
UNCOV 51	for path in bootstrap_options.pythonpath:	×
52	if path not in sys.path:	×
53	sys.path.append(path)	×
54
55	# Resolve the actual Python code for any plugins. `sys.path` is modified as a side effect if
56	# plugins were configured.
UNCOV 57	backends_requirements = _collect_backends_requirements(bootstrap_options.backend_packages)	×
UNCOV 58	plugin_resolver.resolve(options_bootstrapper, env, backends_requirements)	×
59
60	# Load plugins and backends.
UNCOV 61	return load_backends_and_plugins(	×
62	bootstrap_options.plugins,
63	bootstrap_options.backend_packages,
64	BuildConfiguration.Builder(_pants_ng=bootstrap_options.pants_ng),
65	)
66
67
68	def _collect_backends_requirements(backends: list[str]) -> list[str]:	2✔
69	"""Collects backend package dependencies, in case those are declared in an adjacent
70	requirements.txt. Ignores any loading errors, assuming those will be later on handled by the
71	backends loader.
72
73	:param backends: An list of packages to load v2 backends requirements from.
74	"""
UNCOV 75	requirements = []	×
76
UNCOV 77	for backend_package in backends:	×
78	try:	×
79	backend_package_spec = importlib.util.find_spec(backend_package)	×
80	except ModuleNotFoundError:	×
81	continue	×
82
83	if backend_package_spec is None:	×
84	continue	×
85
86	if backend_package_spec.origin is None:	×
87	logger.warning(	×
88	f"Can not check requirements for backend: '{backend_package}'. A __init__.py file is probably missing."
89	)
90	continue	×
91
92	requirements_txt_file_path = Path(backend_package_spec.origin).parent.joinpath(	×
93	"requirements.txt"
94	)
95	if requirements_txt_file_path.exists():	×
96	content = requirements_txt_file_path.read_text()	×
97	backend_package_requirements = [	×
98	str(r)
99	for r in parse_requirements_file(content, rel_path=str(requirements_txt_file_path))
100	]
101	requirements.extend(backend_package_requirements)	×
102
UNCOV 103	return requirements	×
104
105
106	def create_bootstrap_scheduler(	2✔
107	options_bootstrapper: OptionsBootstrapper, executor: PyExecutor
108	) -> BootstrapScheduler:
UNCOV 109	bc_builder = BuildConfiguration.Builder()	×
110	# To load plugins, we only need access to the Python/PEX rules.
UNCOV 111	load_build_configuration_from_source(bc_builder, ["pants.backend.python"])	×
112	# And to plugin-loading-specific rules.
UNCOV 113	bc_builder.register_rules("_dummy_for_bootstrapping_", plugin_resolver_rules())	×
114	# We allow unrecognized options to defer any option error handling until post-bootstrap.
UNCOV 115	bc_builder.allow_unknown_options()	×
UNCOV 116	return BootstrapScheduler(	×
117	EngineInitializer.setup_graph(
118	options_bootstrapper.bootstrap_options.for_global_scope(),
119	bc_builder.create(),
120	DynamicRemoteOptions.disabled(),
121	executor,
122	is_bootstrap=True,
123	).scheduler
124	)
125
126
127	class OptionsInitializer:	2✔
128	"""Initializes BuildConfiguration and Options instances given an OptionsBootstrapper.
129
130	NB: Although this constructor takes an instance of the OptionsBootstrapper, it is
131	used only to construct a "bootstrap" Scheduler: actual calls to resolve plugins use a
132	per-request instance of the OptionsBootstrapper, which might request different plugins.
133
134	TODO: We would eventually like to use the bootstrap Scheduler to construct the
135	OptionsBootstrapper as well, but for now we do the opposite thing, and the Scheduler is
136	used only to resolve plugins.
137	see https://github.com/pantsbuild/pants/pull/11568
138	"""
139
140	def __init__(	2✔
141	self,
142	options_bootstrapper: OptionsBootstrapper,
143	executor: PyExecutor,
144	) -> None:
UNCOV 145	self._bootstrap_scheduler = create_bootstrap_scheduler(options_bootstrapper, executor)	×
UNCOV 146	self._plugin_resolver = PluginResolver(self._bootstrap_scheduler)	×
147
148	def build_config(	2✔
149	self,
150	options_bootstrapper: OptionsBootstrapper,
151	env: CompleteEnvironmentVars,
152	) -> BuildConfiguration:
UNCOV 153	return _initialize_build_configuration(self._plugin_resolver, options_bootstrapper, env)	×
154
155	def options(	2✔
156	self,
157	options_bootstrapper: OptionsBootstrapper,
158	env: CompleteEnvironmentVars,
159	build_config: BuildConfiguration,
160	union_membership: UnionMembership,
161	*,
162	raise_: bool,
163	) -> Options:
UNCOV 164	with self.handle_unknown_flags(options_bootstrapper, env, raise_=raise_):	×
UNCOV 165	return options_bootstrapper.full_options(	×
166	build_config.known_scope_infos, union_membership, build_config.allow_unknown_options
167	)
168
169	@contextmanager	2✔
170	def handle_unknown_flags(	2✔
171	self,
172	options_bootstrapper: OptionsBootstrapper,
173	env: CompleteEnvironmentVars,
174	*,
175	raise_: bool,
176	) -> Iterator[None]:
177	"""If there are any unknown flags, print "Did you mean?" and possibly error."""
UNCOV 178	try:	×
UNCOV 179	yield	×
180	except UnknownFlagsError as err:	×
181	build_config = _initialize_build_configuration(	×
182	self._plugin_resolver, options_bootstrapper, env
183	)
184	# We need an options instance in order to get "did you mean" suggestions, but we know
185	# there are bad flags in the args, so we generate options with no flags.
186	no_arg_bootstrapper = dataclasses.replace(	×
187	options_bootstrapper, args=("dummy_first_arg",)
188	)
189	options = no_arg_bootstrapper.full_options(	×
190	build_config.known_scope_infos,
191	union_membership=UnionMembership.empty(),
192	)
193	FlagErrorHelpPrinter(options).handle_unknown_flags(err)	×
194	if raise_:	×
195	raise err	×

pantsbuild / pants / 26080722777

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