18517631058

Committed 15 Oct 2025 04:18AM UTC coverage: 69.207% (-11.1%) from 80.267%

Build # 18517631058

Build Type

Pull #22745

github

Committed by

web-flow

Commit Message

Merge 642a76ca1 into 99919310e

Pull Request Pull Request #22745: [windows] Add windows support in the stdio crate.

Run Details

53815 of 77759 relevant lines covered (69.21%)

2.42 hits per line

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

28.42

/src/python/pants/backend/docker/utils.py

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

from __future__ import annotations

import difflib
import os.path
import re
from collections.abc import Callable, Iterable, Iterator, Sequence
from fnmatch import fnmatch
from typing import Self, TypeVar

from pants.help.maybe_color import MaybeColor
from pants.util.ordered_set import FrozenOrderedSet

_T = TypeVar("_T", bound="KeyValueSequenceUtil")


image_ref_regexp = re.compile(
    r"""
    ^
    # Optional registry.
    ((?P<registry>[^/:_ ]+:?[^/:_ ]*)/)?
    # Repository.
    (?P<repository>[^:@ \t\n\r\f\v]+)
    # Optionally with `:tag`.
    (:(?P<tag>[^@ ]+))?
    # Optionally with `@digest`.
    (@(?P<digest>\S+))?
    $
    """,
    re.VERBOSE,
)


class KeyValueSequenceUtil(FrozenOrderedSet[str]):
    @classmethod
    def from_strings(cls: type[_T], *strings: str, duplicates_must_match: bool = False) -> _T:
        """Takes all `KEY`/`KEY=VALUE` strings and dedupes by `KEY`.

        The last seen `KEY` wins in case of duplicates, unless `duplicates_must_match` is `True`, in
        which case all `VALUE`s must be equal, if present.
        """

        key_to_entry_and_value: dict[str, tuple[str, str | None]] = {}
        for entry in strings:
            key, has_value, value = entry.partition("=")
            if not duplicates_must_match:
                # Note that last entry with the same key wins.
                key_to_entry_and_value[key] = (entry, value if has_value else None)
            else:
                prev_entry, prev_value = key_to_entry_and_value.get(key, (None, None))
                if prev_entry is None:
                    # Not seen before.
                    key_to_entry_and_value[key] = (entry, value if has_value else None)
                elif not has_value:
                    # Seen before, no new value, so keep existing.
                    pass
                elif prev_value is None:
                    # Update value.
                    key_to_entry_and_value[key] = (entry, value)
                elif prev_value != value:
                    # Seen before with a different value.
                    raise ValueError(
                        f"{cls.__name__}: duplicated {key!r} with different values: "
                        f"{prev_value!r} != {value!r}."
                    )

        deduped_entries = sorted(
            entry_and_value[0] for entry_and_value in key_to_entry_and_value.values()
        )
        return cls(FrozenOrderedSet(deduped_entries))

    @classmethod
    def from_dict(cls, value: dict[str, str | None]) -> Self:
        def encode_kv(k: str, v: str | None) -> str:
            if v is None:
                return k
            return "=".join((k, v))

        return cls(FrozenOrderedSet(sorted(encode_kv(k, v) for k, v in value.items())))

    def to_dict(
        self, default: Callable[[str], str | None] = lambda x: None
    ) -> dict[str, str | None]:
        return {
            key: value if has_value else default(key)
            for key, has_value, value in [pair.partition("=") for pair in self]
        }


def suggest_renames(
    tentative_paths: Iterable[str], actual_files: Sequence[str], actual_dirs: Sequence[str]
) -> Iterator[tuple[str, str]]:
    """Return each pair of `tentative_paths` matched to the best possible match of `actual_paths`
    that are not an exact match.

    A pair of `(tentative_path, "")` means there were no possible match to find in the
    `actual_paths`, while a pair of `("", actual_path)` indicates a file in the build context that
    is not taking part in any `COPY` instruction.
    """

    actual_paths = (*actual_files, *actual_dirs)
    referenced: dict[str, set[str] | bool] = {}

    def reference(path: str) -> None:
        """Track which actual files has been referenced either explicitly by a tentative path, or as
        a suggested rename."""
        if path in actual_dirs:
            referenced[path] = True
        else:
            dirname = os.path.dirname(path)
            refs = referenced.setdefault(dirname, set())
            if isinstance(refs, set):
                refs.add(path)

        # Recalculate possible matches, to avoid suggesting the same file twice.
        nonlocal actual_paths
        actual_paths = tuple(get_unreferenced(actual_files, actual_dirs))

    def is_referenced(path: str, dirname: str | None = None) -> bool:
        """Check the list of referenced files to see if `path` has been flagged.

        Walks up the directory tree in case there is a recursive flag on one of the parent
        directories.
        """
        if dirname is None:
            dirname = os.path.dirname(path)
        refs = referenced.get(dirname, set())
        if isinstance(refs, bool):
            return refs
        if path in refs:
            return True
        parentdir = os.path.dirname(dirname)
        if parentdir:
            return is_referenced(path, parentdir)
        return False

    def get_unreferenced(files: Sequence[str] = (), dirs: Sequence[str] = ()) -> Iterator[str]:
        unreferenced_files = tuple(path for path in files if not is_referenced(path))
        yield from unreferenced_files
        for path in dirs:
            if not any(filename.startswith(path + "/") for filename in unreferenced_files):
                # Skip paths where we don't have any unreferenced files any longer.
                continue
            if not is_referenced(path, path):
                yield path

    def get_matches(path: str) -> tuple[str, ...]:
        is_pattern = any(all(c in path for c in cs) for cs in ["*", "?", "[]"])
        if not is_pattern:
            return (path,) if path in actual_paths else ()
        #
        # NOTICE: There is a slight difference in the pattern syntax used for the Dockerfile `COPY`
        # instruction, than what is implemented by the `fnmatch` function in Python.
        # https://docs.docker.com/engine/reference/builder/#copy which is implemented using
        # https://golang.org/pkg/path/filepath#Match compared to
        # https://docs.python.org/3/library/fnmatch.html#fnmatch.fnmatch
        #
        # For best experience when using globs, this should be addressed, but for now, I'll settle
        # for a "close enough" approximation to get this moving.
        return tuple(p for p in actual_paths if fnmatch(p, path))

    # Go over exact matches first, so we don't target them as possible matches for renames.
    unmatched_paths = set()
    for path in tentative_paths:
        matches = get_matches(path)
        for match in matches:
            reference(match)
        if not matches:
            unmatched_paths.add(path)

    # List unknown files, possibly with a rename suggestion.
    for path in sorted(unmatched_paths):
        for suggestion in difflib.get_close_matches(path, actual_paths, n=1, cutoff=0.1):
            # Suggest rename to match what files there are.
            reference(suggestion)
            yield path, suggestion
            break
        else:
            # No match for this path.
            yield path, ""

    # List unused files.
    for path in sorted(get_unreferenced(actual_files)):
        yield "", path


def format_rename_suggestion(src_path: str, dst_path: str, *, colors: bool) -> str:
    """Given two paths, formats a line showing what to change in `src_path` to get to `dst_path`."""
    color = MaybeColor(colors)
    rem = color.maybe_red(src_path)
    add = color.maybe_green(dst_path)
    return f"{rem} => {add}"

1	# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	5✔
5
6	import difflib	5✔
7	import os.path	5✔
8	import re	5✔
9	from collections.abc import Callable, Iterable, Iterator, Sequence	5✔
10	from fnmatch import fnmatch	5✔
11	from typing import Self, TypeVar	5✔
12
13	from pants.help.maybe_color import MaybeColor	5✔
14	from pants.util.ordered_set import FrozenOrderedSet	5✔
15
16	_T = TypeVar("_T", bound="KeyValueSequenceUtil")	5✔
17
18
19	image_ref_regexp = re.compile(	5✔
20	r"""
21	^
22	# Optional registry.
23	((?P<registry>[^/:_ ]+:?[^/:_ ]*)/)?
24	# Repository.
25	(?P<repository>[^:@ \t\n\r\f\v]+)
26	# Optionally with `:tag`.
27	(:(?P<tag>[^@ ]+))?
28	# Optionally with `@digest`.
29	(@(?P<digest>\S+))?
30	$
31	""",
32	re.VERBOSE,
33	)
34
35
36	class KeyValueSequenceUtil(FrozenOrderedSet[str]):	5✔
37	@classmethod	5✔
38	def from_strings(cls: type[_T], *strings: str, duplicates_must_match: bool = False) -> _T:	5✔
39	"""Takes all `KEY`/`KEY=VALUE` strings and dedupes by `KEY`.
40
41	The last seen `KEY` wins in case of duplicates, unless `duplicates_must_match` is `True`, in
42	which case all `VALUE`s must be equal, if present.
43	"""
44
45	key_to_entry_and_value: dict[str, tuple[str, str \| None]] = {}	1✔
46	for entry in strings:	1✔
47	key, has_value, value = entry.partition("=")	1✔
48	if not duplicates_must_match:	1✔
49	# Note that last entry with the same key wins.
50	key_to_entry_and_value[key] = (entry, value if has_value else None)	1✔
51	else:
52	prev_entry, prev_value = key_to_entry_and_value.get(key, (None, None))	×
53	if prev_entry is None:	×
54	# Not seen before.
55	key_to_entry_and_value[key] = (entry, value if has_value else None)	×
56	elif not has_value:	×
57	# Seen before, no new value, so keep existing.
58	pass	×
59	elif prev_value is None:	×
60	# Update value.
61	key_to_entry_and_value[key] = (entry, value)	×
62	elif prev_value != value:	×
63	# Seen before with a different value.
64	raise ValueError(	×
65	f"{cls.__name__}: duplicated {key!r} with different values: "
66	f"{prev_value!r} != {value!r}."
67	)
68
69	deduped_entries = sorted(	1✔
70	entry_and_value[0] for entry_and_value in key_to_entry_and_value.values()
71	)
72	return cls(FrozenOrderedSet(deduped_entries))	1✔
73
74	@classmethod	5✔
75	def from_dict(cls, value: dict[str, str \| None]) -> Self:	5✔
76	def encode_kv(k: str, v: str \| None) -> str:	×
77	if v is None:	×
78	return k	×
79	return "=".join((k, v))	×
80
81	return cls(FrozenOrderedSet(sorted(encode_kv(k, v) for k, v in value.items())))	×
82
83	def to_dict(	5✔
84	self, default: Callable[[str], str \| None] = lambda x: None
85	) -> dict[str, str \| None]:
86	return {	1✔
87	key: value if has_value else default(key)
88	for key, has_value, value in [pair.partition("=") for pair in self]
89	}
90
91
92	def suggest_renames(	5✔
93	tentative_paths: Iterable[str], actual_files: Sequence[str], actual_dirs: Sequence[str]
94	) -> Iterator[tuple[str, str]]:
95	"""Return each pair of `tentative_paths` matched to the best possible match of `actual_paths`
96	that are not an exact match.
97
98	A pair of `(tentative_path, "")` means there were no possible match to find in the
99	`actual_paths`, while a pair of `("", actual_path)` indicates a file in the build context that
100	is not taking part in any `COPY` instruction.
101	"""
102
103	actual_paths = (actual_files, actual_dirs)	×
104	referenced: dict[str, set[str] \| bool] = {}	×
105
106	def reference(path: str) -> None:	×
107	"""Track which actual files has been referenced either explicitly by a tentative path, or as
108	a suggested rename."""
109	if path in actual_dirs:	×
110	referenced[path] = True	×
111	else:
112	dirname = os.path.dirname(path)	×
113	refs = referenced.setdefault(dirname, set())	×
114	if isinstance(refs, set):	×
115	refs.add(path)	×
116
117	# Recalculate possible matches, to avoid suggesting the same file twice.
118	nonlocal actual_paths
119	actual_paths = tuple(get_unreferenced(actual_files, actual_dirs))	×
120
121	def is_referenced(path: str, dirname: str \| None = None) -> bool:	×
122	"""Check the list of referenced files to see if `path` has been flagged.
123
124	Walks up the directory tree in case there is a recursive flag on one of the parent
125	directories.
126	"""
127	if dirname is None:	×
128	dirname = os.path.dirname(path)	×
129	refs = referenced.get(dirname, set())	×
130	if isinstance(refs, bool):	×
131	return refs	×
132	if path in refs:	×
133	return True	×
134	parentdir = os.path.dirname(dirname)	×
135	if parentdir:	×
136	return is_referenced(path, parentdir)	×
137	return False	×
138
139	def get_unreferenced(files: Sequence[str] = (), dirs: Sequence[str] = ()) -> Iterator[str]:	×
140	unreferenced_files = tuple(path for path in files if not is_referenced(path))	×
141	yield from unreferenced_files	×
142	for path in dirs:	×
143	if not any(filename.startswith(path + "/") for filename in unreferenced_files):	×
144	# Skip paths where we don't have any unreferenced files any longer.
145	continue	×
146	if not is_referenced(path, path):	×
147	yield path	×
148
149	def get_matches(path: str) -> tuple[str, ...]:	×
150	is_pattern = any(all(c in path for c in cs) for cs in ["*", "?", "[]"])	×
151	if not is_pattern:	×
152	return (path,) if path in actual_paths else ()	×
153	#
154	# NOTICE: There is a slight difference in the pattern syntax used for the Dockerfile `COPY`
155	# instruction, than what is implemented by the `fnmatch` function in Python.
156	# https://docs.docker.com/engine/reference/builder/#copy which is implemented using
157	# https://golang.org/pkg/path/filepath#Match compared to
158	# https://docs.python.org/3/library/fnmatch.html#fnmatch.fnmatch
159	#
160	# For best experience when using globs, this should be addressed, but for now, I'll settle
161	# for a "close enough" approximation to get this moving.
162	return tuple(p for p in actual_paths if fnmatch(p, path))	×
163
164	# Go over exact matches first, so we don't target them as possible matches for renames.
165	unmatched_paths = set()	×
166	for path in tentative_paths:	×
167	matches = get_matches(path)	×
168	for match in matches:	×
169	reference(match)	×
170	if not matches:	×
171	unmatched_paths.add(path)	×
172
173	# List unknown files, possibly with a rename suggestion.
174	for path in sorted(unmatched_paths):	×
175	for suggestion in difflib.get_close_matches(path, actual_paths, n=1, cutoff=0.1):	×
176	# Suggest rename to match what files there are.
177	reference(suggestion)	×
178	yield path, suggestion	×
179	break	×
180	else:
181	# No match for this path.
182	yield path, ""	×
183
184	# List unused files.
185	for path in sorted(get_unreferenced(actual_files)):	×
186	yield "", path	×
187
188
189	def format_rename_suggestion(src_path: str, dst_path: str, *, colors: bool) -> str:	5✔
190	"""Given two paths, formats a line showing what to change in `src_path` to get to `dst_path`."""
191	color = MaybeColor(colors)	×
192	rem = color.maybe_red(src_path)	×
193	add = color.maybe_green(dst_path)	×
194	return f"{rem} => {add}"	×

pantsbuild / pants / 18517631058

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