• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

pantsbuild / pants / 19015773527

02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%
19015773527

Pull #22816

github

web-flow
Merge a12d75757 into 6c024e162
Pull Request #22816: Update Pants internal Python to 3.14

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

21.02
/src/python/pants/option/native_options.py
1
# Copyright 2024 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 inspect
1✔
7
import logging
1✔
8
import shlex
1✔
9
from collections.abc import Mapping, Sequence
1✔
10
from enum import Enum
1✔
11
from pathlib import Path
1✔
12
from typing import Any
1✔
13

14
from pants.base.build_environment import get_buildroot
1✔
15
from pants.engine.fs import FileContent
1✔
16
from pants.engine.internals import native_engine
1✔
17
from pants.engine.internals.native_engine import PyConfigSource, PyGoalInfo, PyPantsCommand
1✔
18
from pants.option.custom_types import _flatten_shlexed_list, dir_option, file_option, shell_str
1✔
19
from pants.option.errors import BooleanOptionNameWithNo, OptionsError, ParseError
1✔
20
from pants.option.option_types import OptionInfo
1✔
21
from pants.option.ranked_value import Rank
1✔
22
from pants.option.scope import GLOBAL_SCOPE
1✔
23
from pants.util.strutil import get_strict_env, softwrap
1✔
24

25
logger = logging.getLogger()
1✔
26

27

28
def parse_dest(option_info: OptionInfo) -> str:
1✔
29
    """Return the dest for an option registration.
30

31
    If an explicit `dest` is specified, returns that and otherwise derives a default from the
32
    option flags where '--foo-bar' -> 'foo_bar' and '-x' -> 'x'.
33

34
    The dest is used for:
35
      - The name of the field containing the option value.
36
      - The key in the config file.
37
      - Computing the name of the env var used to set the option name.
38
    """
UNCOV
39
    dest = option_info.kwargs.get("dest")
×
UNCOV
40
    if dest:
×
UNCOV
41
        return str(dest)
×
42
    # No explicit dest, so compute one based on the first long arg, or the short arg
43
    # if that's all there is.
UNCOV
44
    arg = next((a for a in option_info.args if a.startswith("--")), option_info.args[0])
×
UNCOV
45
    return arg.lstrip("-").replace("-", "_")
×
46

47

48
class NativeOptionParser:
1✔
49
    """A Python wrapper around the Rust options parser."""
50

51
    int_to_rank = [
1✔
52
        Rank.NONE,
53
        Rank.HARDCODED,
54
        Rank.CONFIG_DEFAULT,
55
        Rank.CONFIG,
56
        Rank.ENVIRONMENT,
57
        Rank.FLAG,
58
    ]
59

60
    def __init__(
1✔
61
        self,
62
        args: Sequence[str] | None,
63
        env: Mapping[str, str],
64
        config_sources: Sequence[FileContent] | None,
65
        allow_pantsrc: bool,
66
        include_derivation: bool,
67
        known_scopes_to_flags: dict[str, frozenset[str]],
68
        known_goals: Sequence[PyGoalInfo],
69
    ):
70
        # Remember these args so this object can clone itself in with_derivation() below.
UNCOV
71
        (
×
72
            self._args,
73
            self._env,
74
            self._config_sources,
75
            self._allow_pantsrc,
76
            self._known_scopes_to_flags,
77
            self._known_goals,
78
        ) = (
79
            args,
80
            env,
81
            config_sources,
82
            allow_pantsrc,
83
            known_scopes_to_flags,
84
            known_goals,
85
        )
86

UNCOV
87
        py_config_sources = (
×
88
            None
89
            if config_sources is None
90
            else [PyConfigSource(cs.path, cs.content) for cs in config_sources]
91
        )
UNCOV
92
        self._native_parser = native_engine.PyOptionParser(
×
93
            args,
94
            dict(get_strict_env(env, logger)),
95
            py_config_sources,
96
            allow_pantsrc,
97
            include_derivation,
98
            known_scopes_to_flags,
99
            known_goals,
100
        )
101

102
        # (type, member_type) -> native get for that type.
UNCOV
103
        self._getter_by_type = {
×
104
            (bool, None): self._native_parser.get_bool,
105
            (int, None): self._native_parser.get_int,
106
            (float, None): self._native_parser.get_float,
107
            (str, None): self._native_parser.get_string,
108
            (list, bool): self._native_parser.get_bool_list,
109
            (list, int): self._native_parser.get_int_list,
110
            (list, float): self._native_parser.get_float_list,
111
            (list, str): self._native_parser.get_string_list,
112
            (dict, None): self._native_parser.get_dict,
113
        }
114

115
    def with_derivation(self) -> NativeOptionParser:
1✔
116
        """Return a clone of this object but with value derivation enabled."""
117
        # We may be able to get rid of this method once we remove the legacy parser entirely.
118
        # For now it is convenient to allow the help mechanism to get derivations via the
119
        # existing Options object, which otherwise does not need derivations.
UNCOV
120
        return NativeOptionParser(
×
121
            args=None if self._args is None else tuple(self._args),
122
            env=dict(self._env),
123
            config_sources=None if self._config_sources is None else tuple(self._config_sources),
124
            allow_pantsrc=self._allow_pantsrc,
125
            include_derivation=True,
126
            known_scopes_to_flags=self._known_scopes_to_flags,
127
            known_goals=self._known_goals,
128
        )
129

130
    def get_value(self, *, scope: str, option_info: OptionInfo) -> tuple[Any, Rank]:
1✔
UNCOV
131
        val, rank, _ = self._get_value_and_derivation(scope, option_info)
×
UNCOV
132
        return (val, rank)
×
133

134
    def get_derivation(
1✔
135
        self,
136
        scope: str,
137
        option_info: OptionInfo,
138
    ) -> list[tuple[Any, Rank, str | None]]:
UNCOV
139
        _, _, derivation = self._get_value_and_derivation(scope, option_info)
×
UNCOV
140
        return derivation
×
141

142
    def _get_value_and_derivation(
1✔
143
        self,
144
        scope: str,
145
        option_info: OptionInfo,
146
    ) -> tuple[Any, Rank, list[tuple[Any, Rank, str | None]]]:
UNCOV
147
        return self._get(
×
148
            scope=scope,
149
            dest=parse_dest(option_info),
150
            flags=option_info.args,
151
            default=option_info.kwargs.get("default"),
152
            option_type=option_info.kwargs.get("type"),
153
            member_type=option_info.kwargs.get("member_type"),
154
            choices=option_info.kwargs.get("choices"),
155
            passthrough=option_info.kwargs.get("passthrough"),
156
        )
157

158
    def _get(
1✔
159
        self,
160
        *,
161
        scope,
162
        dest,
163
        flags,
164
        default,
165
        option_type,
166
        member_type=None,
167
        choices=None,
168
        passthrough=False,
169
    ) -> tuple[Any, Rank, list[tuple[Any, Rank, str | None]]]:
UNCOV
170
        def scope_str() -> str:
×
UNCOV
171
            return "global scope" if scope == GLOBAL_SCOPE else f"scope '{scope}'"
×
172

UNCOV
173
        def is_enum(typ):
×
174
            # TODO: When we switch to Python 3.11, use: return isinstance(typ, EnumType)
UNCOV
175
            return inspect.isclass(typ) and issubclass(typ, Enum)
×
176

UNCOV
177
        def apply_callable(callable_type, val_str):
×
UNCOV
178
            try:
×
UNCOV
179
                return callable_type(val_str)
×
UNCOV
180
            except (TypeError, ValueError) as e:
×
UNCOV
181
                if is_enum(callable_type):
×
UNCOV
182
                    choices_str = ", ".join(f"{choice.value}" for choice in callable_type)
×
UNCOV
183
                    raise ParseError(f"Invalid choice '{val_str}'. Choose from: {choices_str}")
×
184
                raise ParseError(
×
185
                    f"Error applying type '{callable_type.__name__}' to option value '{val_str}': {e}"
186
                )
187

188
        # '--foo.bar-baz' -> ['foo', 'bar', 'baz']
UNCOV
189
        name_parts = flags[-1][2:].replace(".", "-").split("-")
×
UNCOV
190
        switch = flags[0][1:] if len(flags) > 1 else None  # '-d' -> 'd'
×
UNCOV
191
        option_id = native_engine.PyOptionId(*name_parts, scope=scope or "GLOBAL", switch=switch)
×
192

UNCOV
193
        rust_option_type = option_type
×
UNCOV
194
        rust_member_type = member_type
×
195

UNCOV
196
        if option_type is bool:
×
UNCOV
197
            if name_parts[0] == "no":
×
UNCOV
198
                raise BooleanOptionNameWithNo(scope, dest)
×
UNCOV
199
        elif option_type is dict:
×
200
            # The Python code allows registering default=None for dicts/lists, and forces it to
201
            # an empty dict/list at registration. Since here we only have access to what the user
202
            # provided, we do the same.
UNCOV
203
            if default is None:
×
UNCOV
204
                default = {}
×
UNCOV
205
            elif isinstance(default, str):
×
UNCOV
206
                default = eval(default)
×
UNCOV
207
        elif option_type is list:
×
UNCOV
208
            if default is None:
×
UNCOV
209
                default = []
×
UNCOV
210
            if member_type is None:
×
UNCOV
211
                member_type = rust_member_type = str
×
212

UNCOV
213
            if member_type == shell_str:
×
UNCOV
214
                rust_member_type = str
×
UNCOV
215
                if isinstance(default, str):
×
UNCOV
216
                    default = shlex.split(default)
×
UNCOV
217
            elif is_enum(member_type):
×
UNCOV
218
                rust_member_type = str
×
UNCOV
219
                default = [x.value for x in default]
×
UNCOV
220
            elif inspect.isfunction(rust_member_type):
×
UNCOV
221
                rust_member_type = str
×
UNCOV
222
            elif rust_member_type != str and isinstance(default, str):
×
UNCOV
223
                default = eval(default)
×
UNCOV
224
        elif is_enum(option_type):
×
UNCOV
225
            if default is not None:
×
UNCOV
226
                default = default.value
×
UNCOV
227
                rust_option_type = type(default)
×
228
            else:
UNCOV
229
                rust_option_type = str
×
UNCOV
230
        elif option_type not in {bool, int, float, str}:
×
231
            # For enum and other specialized types.
UNCOV
232
            rust_option_type = str
×
UNCOV
233
            if default is not None:
×
UNCOV
234
                default = str(default)
×
235

UNCOV
236
        getter = self._getter_by_type.get((rust_option_type, rust_member_type))
×
UNCOV
237
        if getter is None:
×
238
            suffix = f" with member type {rust_member_type}" if rust_option_type is list else ""
×
239
            raise OptionsError(f"Unsupported type: {rust_option_type}{suffix}")
×
240

UNCOV
241
        val, rank_int, derivation = getter(option_id, default)  # type:ignore
×
UNCOV
242
        rank = self.int_to_rank[rank_int]
×
243

UNCOV
244
        def process_value(v):
×
UNCOV
245
            if option_type is list:
×
UNCOV
246
                if member_type == shell_str:
×
UNCOV
247
                    v = _flatten_shlexed_list(v)
×
UNCOV
248
                elif callable(member_type):
×
UNCOV
249
                    v = [apply_callable(member_type, x) for x in v]
×
UNCOV
250
                if passthrough:
×
UNCOV
251
                    v += self._native_parser.get_command().passthru() or []
×
UNCOV
252
            elif callable(option_type):
×
UNCOV
253
                v = apply_callable(option_type, v)
×
UNCOV
254
            return v
×
255

UNCOV
256
        if derivation:
×
UNCOV
257
            derivation = [(process_value(v), self.int_to_rank[r], d) for (v, r, d) in derivation]
×
258

UNCOV
259
        if val is not None:
×
UNCOV
260
            val = process_value(val)
×
261

262
            # Validate the value.
263

UNCOV
264
            def check_scalar_value(val, choices):
×
UNCOV
265
                if choices is None and is_enum(option_type):
×
UNCOV
266
                    choices = list(option_type)
×
UNCOV
267
                if choices is not None and val not in choices:
×
UNCOV
268
                    raise ParseError(
×
269
                        softwrap(
270
                            f"""
271
                            `{val}` is not an allowed value for option {dest} in {scope_str()}.
272
                            Must be one of: {choices}
273
                            """
274
                        )
275
                    )
UNCOV
276
                elif option_type == file_option:
×
UNCOV
277
                    check_file_exists(val, dest, scope_str())
×
UNCOV
278
                elif option_type == dir_option:
×
279
                    check_dir_exists(val, dest, scope_str())
×
280

UNCOV
281
            if isinstance(val, list):
×
UNCOV
282
                for component in val:
×
UNCOV
283
                    check_scalar_value(component, choices)
×
UNCOV
284
                if is_enum(member_type) and len(val) != len(set(val)):
×
UNCOV
285
                    raise ParseError(f"Duplicate enum values specified in list: {val}")
×
UNCOV
286
            elif isinstance(val, dict):
×
UNCOV
287
                for component in val.values():
×
UNCOV
288
                    check_scalar_value(component, choices)
×
289
            else:
UNCOV
290
                check_scalar_value(val, choices)
×
291

UNCOV
292
        return (val, rank, derivation)
×
293

294
    def get_command(self) -> PyPantsCommand:
1✔
UNCOV
295
        return self._native_parser.get_command()
×
296

297
    def get_unconsumed_flags(self) -> dict[str, tuple[str, ...]]:
1✔
UNCOV
298
        return {k: tuple(v) for k, v in self._native_parser.get_unconsumed_flags().items()}
×
299

300
    def validate_config(self, valid_keys: dict[str, set[str]]) -> list[str]:
1✔
301
        return self._native_parser.validate_config(valid_keys)
×
302

303

304
def check_file_exists(val: str, dest: str, scope: str) -> None:
1✔
UNCOV
305
    error_prefix = f"File value `{val}` for option `{dest}` in `{scope}`"
×
UNCOV
306
    try:
×
UNCOV
307
        path = Path(val)
×
UNCOV
308
        path_with_buildroot = Path(get_buildroot(), val)
×
309
    except TypeError:
×
310
        raise ParseError(f"{error_prefix} cannot be parsed as a file path.")
×
UNCOV
311
    if not path.is_file() and not path_with_buildroot.is_file():
×
312
        raise ParseError(f"{error_prefix} does not exist.")
×
313

314

315
def check_dir_exists(val: str, dest: str, scope: str) -> None:
1✔
316
    error_prefix = f"Directory value `{val}` for option `{dest}` in `{scope}`"
×
317
    try:
×
318
        path = Path(val)
×
319
        path_with_buildroot = Path(get_buildroot(), val)
×
320
    except TypeError:
×
321
        raise ParseError(f"{error_prefix} cannot be parsed as a directory path.")
×
322
    if not path.is_dir() and not path_with_buildroot.is_dir():
×
323
        raise ParseError(f"{error_prefix} does not exist.")
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc