24145945949

Committed 08 Apr 2026 04:14PM UTC coverage: 82.077% (-10.8%) from 92.91%

Build # 24145945949

Build Type

Pull #23233

github

Committed by

web-flow

Commit Message

Merge 089d98e3c into 9036734c9

Pull Request Pull Request #23233: Introduce a LockfileFormat enum.

Coverage Stats

8 of 11 new or added lines in 4 files covered. (72.73%)

7635 existing lines in 306 files now uncovered.

63732 of 77649 relevant lines covered (82.08%)

2.96 hits per line

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

81.67

/src/python/pants/core/util_rules/stripped_source_files.py

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

from collections import defaultdict
from dataclasses import dataclass

from pants.core.util_rules.source_files import SourceFiles
from pants.core.util_rules.source_files import rules as source_files_rules
from pants.engine.collection import Collection
from pants.engine.engine_aware import EngineAwareParameter
from pants.engine.fs import DigestSubset, MergeDigests, PathGlobs, RemovePrefix, Snapshot
from pants.engine.intrinsics import digest_subset_to_digest, digest_to_snapshot, remove_prefix
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import SourcesPaths
from pants.source.source_root import (
    SourceRootRequest,
    SourceRootsRequest,
    get_source_root,
    get_source_roots,
)
from pants.source.source_root import rules as source_root_rules
from pants.util.dirutil import fast_relpath


@dataclass(frozen=True)
class StrippedSourceFiles:
    """Wrapper for a snapshot of files whose source roots have been stripped."""

    snapshot: Snapshot


@rule
async def strip_source_roots(source_files: SourceFiles) -> StrippedSourceFiles:
    """Removes source roots from a snapshot.

    E.g. `src/python/pants/util/strutil.py` -> `pants/util/strutil.py`.
    """
    if not source_files.snapshot.files:
        return StrippedSourceFiles(source_files.snapshot)

    if source_files.unrooted_files:
        rooted_files = set(source_files.snapshot.files) - set(source_files.unrooted_files)
        rooted_files_snapshot = await digest_to_snapshot(
            **implicitly(DigestSubset(source_files.snapshot.digest, PathGlobs(rooted_files)))
        )
    else:
        rooted_files_snapshot = source_files.snapshot

    source_roots_result = await get_source_roots(
        SourceRootsRequest.for_files(rooted_files_snapshot.files)
    )

    source_roots_to_files = defaultdict(set)
    for f, root in source_roots_result.path_to_root.items():
        source_roots_to_files[root.path].add(str(f))

    if len(source_roots_to_files) == 1:
        source_root = next(iter(source_roots_to_files.keys()))
        if source_root == ".":
            resulting_snapshot = rooted_files_snapshot
        else:
            resulting_snapshot = await digest_to_snapshot(
                **implicitly(RemovePrefix(rooted_files_snapshot.digest, source_root))
            )
    else:
        digest_subsets = await concurrently(
            digest_subset_to_digest(DigestSubset(rooted_files_snapshot.digest, PathGlobs(files)))
            for files in source_roots_to_files.values()
        )
        resulting_digests = await concurrently(
            remove_prefix(RemovePrefix(digest, source_root))
            for digest, source_root in zip(digest_subsets, source_roots_to_files.keys())
        )
        resulting_snapshot = await digest_to_snapshot(**implicitly(MergeDigests(resulting_digests)))

    # Add the unrooted files back in.
    if source_files.unrooted_files:
        unrooted_files_digest = await digest_subset_to_digest(
            DigestSubset(source_files.snapshot.digest, PathGlobs(source_files.unrooted_files))
        )
        resulting_snapshot = await digest_to_snapshot(
            **implicitly(MergeDigests((resulting_snapshot.digest, unrooted_files_digest)))
        )

    return StrippedSourceFiles(resulting_snapshot)


@dataclass(frozen=True)
class StrippedFileName:
    value: str


@dataclass(frozen=True)
class StrippedFileNameRequest(EngineAwareParameter):
    file_path: str

    def debug_hint(self) -> str:
        return self.file_path


@rule
async def strip_file_name(request: StrippedFileNameRequest) -> StrippedFileName:
    source_root = await get_source_root(SourceRootRequest.for_file(request.file_path))
    return StrippedFileName(
        request.file_path
        if source_root.path == "."
        else fast_relpath(request.file_path, source_root.path)
    )


class StrippedSourceFileNames(Collection[str]):
    """The file names from a target's `sources` field, with source roots stripped."""


@rule
async def strip_sources_paths(sources_paths: SourcesPaths) -> StrippedSourceFileNames:
    if not sources_paths.files:
        return StrippedSourceFileNames()
    source_root = await get_source_root(SourceRootRequest.for_file(sources_paths.files[0]))
    if source_root.path == ".":
        return StrippedSourceFileNames(sources_paths.files)
    return StrippedSourceFileNames(fast_relpath(f, source_root.path) for f in sources_paths.files)


def rules():
    return (*collect_rules(), *source_root_rules(), *source_files_rules())

1	# Copyright 2019 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from collections import defaultdict	7✔
5	from dataclasses import dataclass	7✔
6
7	from pants.core.util_rules.source_files import SourceFiles	7✔
8	from pants.core.util_rules.source_files import rules as source_files_rules	7✔
9	from pants.engine.collection import Collection	7✔
10	from pants.engine.engine_aware import EngineAwareParameter	7✔
11	from pants.engine.fs import DigestSubset, MergeDigests, PathGlobs, RemovePrefix, Snapshot	7✔
12	from pants.engine.intrinsics import digest_subset_to_digest, digest_to_snapshot, remove_prefix	7✔
13	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	7✔
14	from pants.engine.target import SourcesPaths	7✔
15	from pants.source.source_root import (	7✔
16	SourceRootRequest,
17	SourceRootsRequest,
18	get_source_root,
19	get_source_roots,
20	)
21	from pants.source.source_root import rules as source_root_rules	7✔
22	from pants.util.dirutil import fast_relpath	7✔
23
24
25	@dataclass(frozen=True)	7✔
26	class StrippedSourceFiles:	7✔
27	"""Wrapper for a snapshot of files whose source roots have been stripped."""
28
29	snapshot: Snapshot
30
31
32	@rule	7✔
33	async def strip_source_roots(source_files: SourceFiles) -> StrippedSourceFiles:	7✔
34	"""Removes source roots from a snapshot.
35
36	E.g. `src/python/pants/util/strutil.py` -> `pants/util/strutil.py`.
37	"""
38	if not source_files.snapshot.files:	7✔
39	return StrippedSourceFiles(source_files.snapshot)	6✔
40
41	if source_files.unrooted_files:	7✔
UNCOV 42	rooted_files = set(source_files.snapshot.files) - set(source_files.unrooted_files)	×
UNCOV 43	rooted_files_snapshot = await digest_to_snapshot(	×
44	**implicitly(DigestSubset(source_files.snapshot.digest, PathGlobs(rooted_files)))
45	)
46	else:
47	rooted_files_snapshot = source_files.snapshot	7✔
48
49	source_roots_result = await get_source_roots(	7✔
50	SourceRootsRequest.for_files(rooted_files_snapshot.files)
51	)
52
53	source_roots_to_files = defaultdict(set)	7✔
54	for f, root in source_roots_result.path_to_root.items():	7✔
55	source_roots_to_files[root.path].add(str(f))	7✔
56
57	if len(source_roots_to_files) == 1:	7✔
58	source_root = next(iter(source_roots_to_files.keys()))	7✔
59	if source_root == ".":	7✔
60	resulting_snapshot = rooted_files_snapshot	7✔
61	else:
62	resulting_snapshot = await digest_to_snapshot(	7✔
63	**implicitly(RemovePrefix(rooted_files_snapshot.digest, source_root))
64	)
65	else:
66	digest_subsets = await concurrently(	2✔
67	digest_subset_to_digest(DigestSubset(rooted_files_snapshot.digest, PathGlobs(files)))
68	for files in source_roots_to_files.values()
69	)
70	resulting_digests = await concurrently(	2✔
71	remove_prefix(RemovePrefix(digest, source_root))
72	for digest, source_root in zip(digest_subsets, source_roots_to_files.keys())
73	)
74	resulting_snapshot = await digest_to_snapshot(**implicitly(MergeDigests(resulting_digests)))	2✔
75
76	# Add the unrooted files back in.
77	if source_files.unrooted_files:	7✔
UNCOV 78	unrooted_files_digest = await digest_subset_to_digest(	×
79	DigestSubset(source_files.snapshot.digest, PathGlobs(source_files.unrooted_files))
80	)
UNCOV 81	resulting_snapshot = await digest_to_snapshot(	×
82	**implicitly(MergeDigests((resulting_snapshot.digest, unrooted_files_digest)))
83	)
84
85	return StrippedSourceFiles(resulting_snapshot)	7✔
86
87
88	@dataclass(frozen=True)	7✔
89	class StrippedFileName:	7✔
90	value: str
91
92
93	@dataclass(frozen=True)	7✔
94	class StrippedFileNameRequest(EngineAwareParameter):	7✔
95	file_path: str
96
97	def debug_hint(self) -> str:	7✔
98	return self.file_path	×
99
100
101	@rule	7✔
102	async def strip_file_name(request: StrippedFileNameRequest) -> StrippedFileName:	7✔
103	source_root = await get_source_root(SourceRootRequest.for_file(request.file_path))	7✔
104	return StrippedFileName(	7✔
105	request.file_path
106	if source_root.path == "."
107	else fast_relpath(request.file_path, source_root.path)
108	)
109
110
111	class StrippedSourceFileNames(Collection[str]):	7✔
112	"""The file names from a target's `sources` field, with source roots stripped."""
113
114
115	@rule	7✔
116	async def strip_sources_paths(sources_paths: SourcesPaths) -> StrippedSourceFileNames:	7✔
UNCOV 117	if not sources_paths.files:	×
UNCOV 118	return StrippedSourceFileNames()	×
UNCOV 119	source_root = await get_source_root(SourceRootRequest.for_file(sources_paths.files[0]))	×
UNCOV 120	if source_root.path == ".":	×
UNCOV 121	return StrippedSourceFileNames(sources_paths.files)	×
UNCOV 122	return StrippedSourceFileNames(fast_relpath(f, source_root.path) for f in sources_paths.files)	×
123
124
125	def rules():	7✔
126	return (collect_rules(), source_root_rules(), *source_files_rules())	7✔

pantsbuild / pants / 24145945949

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