19381742489

Committed 15 Nov 2025 12:52AM UTC coverage: 49.706% (-30.6%) from 80.29%

Build # 19381742489

Build Type

Pull #22890

github

Committed by

web-flow

Commit Message

Merge d961abf79 into 42e1ebd41

Pull Request Pull Request #22890: Updated all python subsystem constraints to 3.14

Run Details

4 of 5 new or added lines in 5 files covered. (80.0%)

14659 existing lines in 485 files now uncovered.

31583 of 63540 relevant lines covered (49.71%)

0.79 hits per line

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

66.67

/src/python/pants/option/custom_types.py

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

from __future__ import annotations

import inspect
import os
import re
import shlex
from collections.abc import Iterable, Sequence
from enum import Enum
from re import Pattern

from pants.option.errors import ParseError
from pants.util.eval import parse_expression
from pants.util.memo import memoized_method
from pants.util.strutil import softwrap


class UnsetBool:
    """A type that can be used as the default value for a bool typed option to indicate un-set.

    In other words, `bool`-typed options with a `default=UnsetBool` that are not explicitly set will
    have the value `None`, enabling a tri-state.

    :API: public
    """

    def __init__(self) -> None:
        raise NotImplementedError(
            "UnsetBool cannot be instantiated. It should only be used as a sentinel type."
        )

    @classmethod
    def coerce_bool(cls, value: type[UnsetBool] | bool | None, default: bool) -> bool:
        if value is None:
            return default
        if value is cls:
            return default
        assert isinstance(value, bool)
        return value


def target_option(s: str) -> str:
    """Same type as 'str', but indicates a single target spec.

    :API: public

    TODO(stuhood): Eagerly convert these to Addresses: see https://rbcommons.com/s/twitter/r/2937/
    """
    return s


def _normalize_directory_separators(s: str) -> str:
    """Coalesce runs of consecutive instances of `os.sep` in `s`, e.g. '//' -> '/' on POSIX.

    The engine will use paths or target addresses either to form globs or to string-match against, and
    including the directory separator '/' multiple times in a row e.g. '//' produces an equivalent
    glob as with a single '/', but produces a different actual string, which will cause the engine to
    fail to glob file paths or target specs correctly.

    TODO: give the engine more control over matching paths so we don't have to sanitize the input!
    """
    return os.path.normpath(s)


def dir_option(s: str) -> str:
    """Same type as 'str', but indicates string represents a directory path.

    :API: public
    """
    return _normalize_directory_separators(s)


def file_option(s: str) -> str:
    """Same type as 'str', but indicates string represents a filepath.

    :API: public
    """
    return _normalize_directory_separators(s)


def dict_with_files_option(s):
    """Same as 'dict', but fingerprints the file contents of any values which are file paths.

    For any value which matches the path of a file on disk, the file path is not fingerprinted -- only
    its contents.

    :API: public
    """
    return DictValueComponent.create(s)


def shell_str(s: str) -> str:
    """A member_type for strings that should be split upon parsing through `shlex.split()`.

    For example, the option value `--foo --bar=val` would be split into `['--foo', '--bar=val']`,
    and then the parser will safely merge this expanded list with any other values defined for the
    option.

    :API: public
    """
    return s


def workspace_path(s: str) -> str:
    """Same type as 'str', but indicates string represents a directory path that is relative to
    either the build root, or a BUILD file if prefix with `./`.

    :API: public
    """
    if s.startswith("/"):
        raise ParseError(
            softwrap(
                f"""
                Invalid value: `{s}`. Expected a relative path, optionally in the form
                `./relative/path` to make it relative to the BUILD files rather than the build root.
                """
            )
        )
    return s


def memory_size(s: str | int | float) -> int:
    """A string that normalizes the suffixes {GiB, MiB, KiB, B} into the number of bytes.

    :API: public
    """
    if isinstance(s, (int, float)):
        return int(s)
    if not s:
        raise ParseError("Missing value.")

    original = s
    s = s.lower().strip()

    try:
        return int(float(s))
    except ValueError:
        pass

    invalid = ParseError(
        softwrap(
            f"""
            Invalid value: `{original}`. Expected either a bare number or a number with one of
            `GiB`, `MiB`, `KiB`, or `B`.
            """
        )
    )

    def convert_to_bytes(power_of_2) -> int:
        try:
            return int(float(s[:-3]) * (2**power_of_2))
        except TypeError:
            raise invalid

    if s.endswith("gib"):
        return convert_to_bytes(30)
    elif s.endswith("mib"):
        return convert_to_bytes(20)
    elif s.endswith("kib"):
        return convert_to_bytes(10)
    elif s.endswith("b"):
        try:
            return int(float(s[:-1]))
        except TypeError:
            raise invalid
    raise invalid


def _convert(val, acceptable_types):
    """Ensure that val is one of the acceptable types, converting it if needed.

    :param val: The value we're parsing (either a string or one of the acceptable types).
    :param acceptable_types: A tuple of expected types for val.
    :returns: The parsed value.
    :raises :class:`pants.options.errors.ParseError`: if there was a problem parsing the val as an
                                                      acceptable type.
    """
    if isinstance(val, acceptable_types):
        return val
    try:
        return parse_expression(val, acceptable_types)
    except ValueError as e:
        raise ParseError(str(e)) from e


def _convert_list(val, member_type, is_enum):
    converted = _convert(val, (list, tuple))
    if not is_enum:
        return converted
    return [item if isinstance(item, member_type) else member_type(item) for item in converted]


def _flatten_shlexed_list(shlexed_args: Sequence[str]) -> list[str]:
    """Convert a list of shlexed args into a flattened list of individual args.

    For example, ['arg1 arg2=foo', '--arg3'] would be converted to ['arg1', 'arg2=foo', '--arg3'].
    """
    return [arg for shlexed_arg in shlexed_args for arg in shlex.split(shlexed_arg)]


class ListValueComponent:
    """A component of the value of a list-typed option.

    One or more instances of this class can be merged to form a list value.

    A component consists of values to append and values to filter while constructing the final list.

    Each component may either replace or modify the preceding component.  So that, e.g., a config
    file can append to and/or filter the default value list, instead of having to repeat most
    of the contents of the default value list.
    """

    REPLACE = "REPLACE"
    MODIFY = "MODIFY"

    # We use a regex to parse the comma-separated lists of modifier expressions (each of which is
    # a list or tuple literal preceded by a + or a -).  Note that these expressions are technically
    # a context-free grammar, but in practice using this regex as a heuristic will work fine. The
    # values that could defeat it are extremely unlikely to be encountered in practice.
    # If we do ever encounter them, we'll have to replace this with a real parser.
    @classmethod
    @memoized_method
    def _get_modifier_expr_re(cls) -> Pattern[str]:
        # Note that the regex consists of a positive lookbehind assertion for a ] or a ),
        # followed by a comma (possibly surrounded by whitespace), followed by a
        # positive lookahead assertion for [ or (.  The lookahead/lookbehind assertions mean that
        # the bracket/paren characters don't get consumed in the split.
        return re.compile(r"(?<=\]|\))\s*,\s*(?=[+-](?:\[|\())")

    @classmethod
    def _split_modifier_expr(cls, s: str) -> list[str]:
        # This check ensures that the first expression (before the first split point) is a modification.
        if s.startswith("+") or s.startswith("-"):
            return cls._get_modifier_expr_re().split(s)
        return [s]

    @classmethod
    def merge(cls, components: Iterable[ListValueComponent]) -> ListValueComponent:
        """Merges components into a single component, applying their actions appropriately.

        This operation is associative:  M(M(a, b), c) == M(a, M(b, c)) == M(a, b, c).
        """
        # Note that action of the merged component is MODIFY until the first REPLACE is encountered.
        # This guarantees associativity.
        action = cls.MODIFY
        appends = []
        filters = []
        for component in components:
            if component._action is cls.REPLACE:
                appends = component._appends
                filters = component._filters
                action = cls.REPLACE
            elif component._action is cls.MODIFY:
                appends.extend(component._appends)
                filters.extend(component._filters)
            else:
                raise ParseError(f"Unknown action for list value: {component._action}")
        return cls(action, appends, filters)

    def __init__(self, action: str, appends: list, filters: list) -> None:
        self._action = action
        self._appends = appends
        self._filters = filters

    @property
    def val(self) -> list:
        ret = list(self._appends)
        for x in self._filters:
            # Note: can't do ret.remove(x) because that only removes the first instance of x.
            ret = [y for y in ret if y != x]
        return ret

    @property
    def action(self):
        return self._action

    @classmethod
    def create(cls, value, member_type=str) -> ListValueComponent:
        """Interpret value as either a list or something to extend another list with.

        Note that we accept tuple literals, but the internal value is always a list.

        :param value: The value to convert.  Can be an instance of ListValueComponent, a list, a tuple,
                      a string representation of a list or tuple (possibly prefixed by + or -
                      indicating modification instead of replacement), or any allowed member_type.
                      May also be a comma-separated sequence of modifications.
        """
        if isinstance(value, cls):  # Ensure idempotency.
            return value

        if isinstance(value, bytes):
            value = value.decode()

        if isinstance(value, str):
            comma_separated_exprs = cls._split_modifier_expr(value)
            if len(comma_separated_exprs) > 1:
                return cls.merge([cls.create(x) for x in comma_separated_exprs])

        action = cls.MODIFY
        appends: Sequence[str] = []
        filters: Sequence[str] = []
        is_enum = inspect.isclass(member_type) and issubclass(member_type, Enum)
        if isinstance(value, (list, tuple)):  # Ensure we can handle list-typed default values.
            action = cls.REPLACE
            appends = value
        elif value.startswith("[") or value.startswith("("):
            action = cls.REPLACE
            appends = _convert_list(value, member_type, is_enum)
        elif value.startswith("+[") or value.startswith("+("):
            appends = _convert_list(value[1:], member_type, is_enum)
        elif value.startswith("-[") or value.startswith("-("):
            filters = _convert_list(value[1:], member_type, is_enum)
        elif is_enum and isinstance(value, str):
            appends = _convert_list([value], member_type, True)
        elif isinstance(value, str):
            appends = [value]
        else:
            appends = _convert(f"[{value}]", list)

        if member_type == shell_str:
            appends = _flatten_shlexed_list(appends)
            filters = _flatten_shlexed_list(filters)

        return cls(action, list(appends), list(filters))

    def __repr__(self) -> str:
        return f"{self._action} +{self._appends} -{self._filters}"


class DictValueComponent:
    """A component of the value of a dict-typed option.

    One or more instances of this class can be merged to form a dict value.

    Each component may either replace or extend the preceding component.  So that, e.g., a config
    file can extend the default value of a dict, instead of having to repeat it.
    """

    REPLACE = "REPLACE"
    EXTEND = "EXTEND"

    @classmethod
    def merge(cls, components: Iterable[DictValueComponent]) -> DictValueComponent:
        """Merges components into a single component, applying their actions appropriately.

        This operation is associative:  M(M(a, b), c) == M(a, M(b, c)) == M(a, b, c).
        """
        # Note that action of the merged component is EXTEND until the first REPLACE is encountered.
        # This guarantees associativity.
        action = cls.EXTEND
        val = {}
        for component in components:
            if component.action is cls.REPLACE:
                val = component.val
                action = cls.REPLACE
            elif component.action is cls.EXTEND:
                val.update(component.val)
            else:
                raise ParseError(f"Unknown action for dict value: {component.action}")
        return cls(action, val)

    def __init__(self, action: str, val: dict) -> None:
        self.action = action
        self.val = val

    @classmethod
    def create(cls, value) -> DictValueComponent:
        """Interpret value as either a dict or something to extend another dict with.

        :param value: The value to convert.  Can be an instance of DictValueComponent, a dict,
                      or a string representation (possibly prefixed by +) of a dict.
        """
        if isinstance(value, bytes):
            value = value.decode()
        if isinstance(value, cls):  # Ensure idempotency.
            action = value.action
            val = value.val
        elif isinstance(value, dict):  # Ensure we can handle dict-typed default values.
            action = cls.REPLACE
            val = value
        elif value.startswith("{"):
            action = cls.REPLACE
            val = _convert(value, dict)
        elif value.startswith("+{"):
            action = cls.EXTEND
            val = _convert(value[1:], dict)
        else:
            raise ParseError(f"Invalid dict value: {value}")
        return cls(action, dict(val))

    def __repr__(self) -> str:
        return f"{self.action} {self.val}"

1	# Copyright 2014 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 inspect	2✔
7	import os	2✔
8	import re	2✔
9	import shlex	2✔
10	from collections.abc import Iterable, Sequence	2✔
11	from enum import Enum	2✔
12	from re import Pattern	2✔
13
14	from pants.option.errors import ParseError	2✔
15	from pants.util.eval import parse_expression	2✔
16	from pants.util.memo import memoized_method	2✔
17	from pants.util.strutil import softwrap	2✔
18
19
20	class UnsetBool:	2✔
21	"""A type that can be used as the default value for a bool typed option to indicate un-set.
22
23	In other words, `bool`-typed options with a `default=UnsetBool` that are not explicitly set will
24	have the value `None`, enabling a tri-state.
25
26	:API: public
27	"""
28
29	def __init__(self) -> None:	2✔
30	raise NotImplementedError(	1✔
31	"UnsetBool cannot be instantiated. It should only be used as a sentinel type."
32	)
33
34	@classmethod	2✔
35	def coerce_bool(cls, value: type[UnsetBool] \| bool \| None, default: bool) -> bool:	2✔
36	if value is None:	×
37	return default	×
38	if value is cls:	×
39	return default	×
40	assert isinstance(value, bool)	×
41	return value	×
42
43
44	def target_option(s: str) -> str:	2✔
45	"""Same type as 'str', but indicates a single target spec.
46
47	:API: public
48
49	TODO(stuhood): Eagerly convert these to Addresses: see https://rbcommons.com/s/twitter/r/2937/
50	"""
UNCOV 51	return s	×
52
53
54	def _normalize_directory_separators(s: str) -> str:	2✔
55	"""Coalesce runs of consecutive instances of `os.sep` in `s`, e.g. '//' -> '/' on POSIX.
56
57	The engine will use paths or target addresses either to form globs or to string-match against, and
58	including the directory separator '/' multiple times in a row e.g. '//' produces an equivalent
59	glob as with a single '/', but produces a different actual string, which will cause the engine to
60	fail to glob file paths or target specs correctly.
61
62	TODO: give the engine more control over matching paths so we don't have to sanitize the input!
63	"""
UNCOV 64	return os.path.normpath(s)	×
65
66
67	def dir_option(s: str) -> str:	2✔
68	"""Same type as 'str', but indicates string represents a directory path.
69
70	:API: public
71	"""
72	return _normalize_directory_separators(s)	×
73
74
75	def file_option(s: str) -> str:	2✔
76	"""Same type as 'str', but indicates string represents a filepath.
77
78	:API: public
79	"""
UNCOV 80	return _normalize_directory_separators(s)	×
81
82
83	def dict_with_files_option(s):	2✔
84	"""Same as 'dict', but fingerprints the file contents of any values which are file paths.
85
86	For any value which matches the path of a file on disk, the file path is not fingerprinted -- only
87	its contents.
88
89	:API: public
90	"""
91	return DictValueComponent.create(s)	×
92
93
94	def shell_str(s: str) -> str:	2✔
95	"""A member_type for strings that should be split upon parsing through `shlex.split()`.
96
97	For example, the option value `--foo --bar=val` would be split into `['--foo', '--bar=val']`,
98	and then the parser will safely merge this expanded list with any other values defined for the
99	option.
100
101	:API: public
102	"""
103	return s	1✔
104
105
106	def workspace_path(s: str) -> str:	2✔
107	"""Same type as 'str', but indicates string represents a directory path that is relative to
108	either the build root, or a BUILD file if prefix with `./`.
109
110	:API: public
111	"""
112	if s.startswith("/"):	1✔
113	raise ParseError(	×
114	softwrap(
115	f"""
116	Invalid value: `{s}`. Expected a relative path, optionally in the form
117	`./relative/path` to make it relative to the BUILD files rather than the build root.
118	"""
119	)
120	)
121	return s	1✔
122
123
124	def memory_size(s: str \| int \| float) -> int:	2✔
125	"""A string that normalizes the suffixes {GiB, MiB, KiB, B} into the number of bytes.
126
127	:API: public
128	"""
129	if isinstance(s, (int, float)):	2✔
130	return int(s)	×
131	if not s:	2✔
132	raise ParseError("Missing value.")	1✔
133
134	original = s	2✔
135	s = s.lower().strip()	2✔
136
137	try:	2✔
138	return int(float(s))	2✔
139	except ValueError:	2✔
140	pass	2✔
141
142	invalid = ParseError(	2✔
143	softwrap(
144	f"""
145	Invalid value: `{original}`. Expected either a bare number or a number with one of
146	`GiB`, `MiB`, `KiB`, or `B`.
147	"""
148	)
149	)
150
151	def convert_to_bytes(power_of_2) -> int:	2✔
152	try:	2✔
153	return int(float(s[:-3]) * (2**power_of_2))	2✔
154	except TypeError:	×
155	raise invalid	×
156
157	if s.endswith("gib"):	2✔
158	return convert_to_bytes(30)	2✔
159	elif s.endswith("mib"):	2✔
160	return convert_to_bytes(20)	2✔
161	elif s.endswith("kib"):	1✔
162	return convert_to_bytes(10)	1✔
163	elif s.endswith("b"):	1✔
164	try:	1✔
165	return int(float(s[:-1]))	1✔
166	except TypeError:	×
167	raise invalid	×
168	raise invalid	1✔
169
170
171	def _convert(val, acceptable_types):	2✔
172	"""Ensure that val is one of the acceptable types, converting it if needed.
173
174	:param val: The value we're parsing (either a string or one of the acceptable types).
175	:param acceptable_types: A tuple of expected types for val.
176	:returns: The parsed value.
177	:raises :class:`pants.options.errors.ParseError`: if there was a problem parsing the val as an
178	acceptable type.
179	"""
180	if isinstance(val, acceptable_types):	1✔
181	return val	×
182	try:	1✔
183	return parse_expression(val, acceptable_types)	1✔
184	except ValueError as e:	×
185	raise ParseError(str(e)) from e	×
186
187
188	def _convert_list(val, member_type, is_enum):	2✔
189	converted = _convert(val, (list, tuple))	1✔
190	if not is_enum:	1✔
191	return converted	1✔
192	return [item if isinstance(item, member_type) else member_type(item) for item in converted]	×
193
194
195	def _flatten_shlexed_list(shlexed_args: Sequence[str]) -> list[str]:	2✔
196	"""Convert a list of shlexed args into a flattened list of individual args.
197
198	For example, ['arg1 arg2=foo', '--arg3'] would be converted to ['arg1', 'arg2=foo', '--arg3'].
199	"""
200	return [arg for shlexed_arg in shlexed_args for arg in shlex.split(shlexed_arg)]	1✔
201
202
203	class ListValueComponent:	2✔
204	"""A component of the value of a list-typed option.
205
206	One or more instances of this class can be merged to form a list value.
207
208	A component consists of values to append and values to filter while constructing the final list.
209
210	Each component may either replace or modify the preceding component. So that, e.g., a config
211	file can append to and/or filter the default value list, instead of having to repeat most
212	of the contents of the default value list.
213	"""
214
215	REPLACE = "REPLACE"	2✔
216	MODIFY = "MODIFY"	2✔
217
218	# We use a regex to parse the comma-separated lists of modifier expressions (each of which is
219	# a list or tuple literal preceded by a + or a -). Note that these expressions are technically
220	# a context-free grammar, but in practice using this regex as a heuristic will work fine. The
221	# values that could defeat it are extremely unlikely to be encountered in practice.
222	# If we do ever encounter them, we'll have to replace this with a real parser.
223	@classmethod	2✔
224	@memoized_method	2✔
225	def _get_modifier_expr_re(cls) -> Pattern[str]:	2✔
226	# Note that the regex consists of a positive lookbehind assertion for a ] or a ),
227	# followed by a comma (possibly surrounded by whitespace), followed by a
228	# positive lookahead assertion for [ or (. The lookahead/lookbehind assertions mean that
229	# the bracket/paren characters don't get consumed in the split.
230	return re.compile(r"(?<=\]\|\))\s,\s(?=[+-](?:\[\|\())")	1✔
231
232	@classmethod	2✔
233	def _split_modifier_expr(cls, s: str) -> list[str]:	2✔
234	# This check ensures that the first expression (before the first split point) is a modification.
235	if s.startswith("+") or s.startswith("-"):	1✔
236	return cls._get_modifier_expr_re().split(s)	1✔
237	return [s]	1✔
238
239	@classmethod	2✔
240	def merge(cls, components: Iterable[ListValueComponent]) -> ListValueComponent:	2✔
241	"""Merges components into a single component, applying their actions appropriately.
242
243	This operation is associative: M(M(a, b), c) == M(a, M(b, c)) == M(a, b, c).
244	"""
245	# Note that action of the merged component is MODIFY until the first REPLACE is encountered.
246	# This guarantees associativity.
247	action = cls.MODIFY	×
248	appends = []	×
249	filters = []	×
250	for component in components:	×
251	if component._action is cls.REPLACE:	×
252	appends = component._appends	×
253	filters = component._filters	×
254	action = cls.REPLACE	×
255	elif component._action is cls.MODIFY:	×
256	appends.extend(component._appends)	×
257	filters.extend(component._filters)	×
258	else:
259	raise ParseError(f"Unknown action for list value: {component._action}")	×
260	return cls(action, appends, filters)	×
261
262	def __init__(self, action: str, appends: list, filters: list) -> None:	2✔
263	self._action = action	1✔
264	self._appends = appends	1✔
265	self._filters = filters	1✔
266
267	@property	2✔
268	def val(self) -> list:	2✔
269	ret = list(self._appends)	1✔
270	for x in self._filters:	1✔
271	# Note: can't do ret.remove(x) because that only removes the first instance of x.
272	ret = [y for y in ret if y != x]	×
273	return ret	1✔
274
275	@property	2✔
276	def action(self):	2✔
277	return self._action	×
278
279	@classmethod	2✔
280	def create(cls, value, member_type=str) -> ListValueComponent:	2✔
281	"""Interpret value as either a list or something to extend another list with.
282
283	Note that we accept tuple literals, but the internal value is always a list.
284
285	:param value: The value to convert. Can be an instance of ListValueComponent, a list, a tuple,
286	a string representation of a list or tuple (possibly prefixed by + or -
287	indicating modification instead of replacement), or any allowed member_type.
288	May also be a comma-separated sequence of modifications.
289	"""
290	if isinstance(value, cls): # Ensure idempotency.	1✔
291	return value	×
292
293	if isinstance(value, bytes):	1✔
294	value = value.decode()	×
295
296	if isinstance(value, str):	1✔
297	comma_separated_exprs = cls._split_modifier_expr(value)	1✔
298	if len(comma_separated_exprs) > 1:	1✔
299	return cls.merge([cls.create(x) for x in comma_separated_exprs])	×
300
301	action = cls.MODIFY	1✔
302	appends: Sequence[str] = []	1✔
303	filters: Sequence[str] = []	1✔
304	is_enum = inspect.isclass(member_type) and issubclass(member_type, Enum)	1✔
305	if isinstance(value, (list, tuple)): # Ensure we can handle list-typed default values.	1✔
306	action = cls.REPLACE	×
307	appends = value	×
308	elif value.startswith("[") or value.startswith("("):	1✔
309	action = cls.REPLACE	1✔
310	appends = _convert_list(value, member_type, is_enum)	1✔
311	elif value.startswith("+[") or value.startswith("+("):	1✔
312	appends = _convert_list(value[1:], member_type, is_enum)	1✔
313	elif value.startswith("-[") or value.startswith("-("):	1✔
314	filters = _convert_list(value[1:], member_type, is_enum)	×
315	elif is_enum and isinstance(value, str):	1✔
316	appends = _convert_list([value], member_type, True)	×
317	elif isinstance(value, str):	1✔
318	appends = [value]	1✔
319	else:
320	appends = _convert(f"[{value}]", list)	×
321
322	if member_type == shell_str:	1✔
UNCOV 323	appends = _flatten_shlexed_list(appends)	×
UNCOV 324	filters = _flatten_shlexed_list(filters)	×
325
326	return cls(action, list(appends), list(filters))	1✔
327
328	def __repr__(self) -> str:	2✔
329	return f"{self._action} +{self._appends} -{self._filters}"	×
330
331
332	class DictValueComponent:	2✔
333	"""A component of the value of a dict-typed option.
334
335	One or more instances of this class can be merged to form a dict value.
336
337	Each component may either replace or extend the preceding component. So that, e.g., a config
338	file can extend the default value of a dict, instead of having to repeat it.
339	"""
340
341	REPLACE = "REPLACE"	2✔
342	EXTEND = "EXTEND"	2✔
343
344	@classmethod	2✔
345	def merge(cls, components: Iterable[DictValueComponent]) -> DictValueComponent:	2✔
346	"""Merges components into a single component, applying their actions appropriately.
347
348	This operation is associative: M(M(a, b), c) == M(a, M(b, c)) == M(a, b, c).
349	"""
350	# Note that action of the merged component is EXTEND until the first REPLACE is encountered.
351	# This guarantees associativity.
352	action = cls.EXTEND	×
353	val = {}	×
354	for component in components:	×
355	if component.action is cls.REPLACE:	×
356	val = component.val	×
357	action = cls.REPLACE	×
358	elif component.action is cls.EXTEND:	×
359	val.update(component.val)	×
360	else:
361	raise ParseError(f"Unknown action for dict value: {component.action}")	×
362	return cls(action, val)	×
363
364	def __init__(self, action: str, val: dict) -> None:	2✔
365	self.action = action	1✔
366	self.val = val	1✔
367
368	@classmethod	2✔
369	def create(cls, value) -> DictValueComponent:	2✔
370	"""Interpret value as either a dict or something to extend another dict with.
371
372	:param value: The value to convert. Can be an instance of DictValueComponent, a dict,
373	or a string representation (possibly prefixed by +) of a dict.
374	"""
375	if isinstance(value, bytes):	1✔
376	value = value.decode()	×
377	if isinstance(value, cls): # Ensure idempotency.	1✔
378	action = value.action	×
379	val = value.val	×
380	elif isinstance(value, dict): # Ensure we can handle dict-typed default values.	1✔
381	action = cls.REPLACE	×
382	val = value	×
383	elif value.startswith("{"):	1✔
384	action = cls.REPLACE	1✔
385	val = _convert(value, dict)	1✔
386	elif value.startswith("+{"):	1✔
387	action = cls.EXTEND	×
388	val = _convert(value[1:], dict)	×
389	else:
390	raise ParseError(f"Invalid dict value: {value}")	1✔
391	return cls(action, dict(val))	1✔
392
393	def __repr__(self) -> str:	2✔
394	return f"{self.action} {self.val}"	×

pantsbuild / pants / 19381742489

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