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

61.04

/src/python/pants/ng/source_partition.py

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

from __future__ import annotations

import os
from dataclasses import dataclass
from pathlib import Path

from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior
from pants.engine.collection import Collection
from pants.engine.fs import GlobExpansionConjunction, PathGlobs, PathMetadataRequest
from pants.engine.internals.native_engine import (
    Digest,
    PathMetadataKind,
    PyNgOptions,
    PyNgOptionsReader,
    PyNgSourcePartition,
)
from pants.engine.internals.session import SessionValues
from pants.engine.intrinsics import path_globs_to_digest, path_globs_to_paths, path_metadata_request
from pants.engine.rules import Rule, _uncacheable_rule, collect_rules, implicitly, rule
from pants.source.source_root import SourceRoot, SourceRootsRequest, get_source_roots
from pants.util.memo import memoized_property


@dataclass(frozen=True)
class SourcePaths:
    """A set of sources, under a single source root."""

    paths: tuple[Path, ...]
    source_root: SourceRoot

    def path_strs(self) -> tuple[str, ...]:
        return tuple(str(path) for path in self.paths)

    def filter_by_suffixes(self, suffixes: tuple[str, ...]) -> SourcePaths:
        suffixes_set = set(suffixes)
        return SourcePaths(
            tuple(path for path in self.paths if path.suffix in suffixes_set),
            self.source_root,
        )


@dataclass(frozen=True)
class CommonDir:
    path: Path | None


@rule
async def find_common_dir(source_paths: SourcePaths) -> CommonDir:
    if not source_paths.paths:  # We don't expect empty SourcePaths, but might as well be robust.
        return CommonDir(None)
    commonpath = os.path.commonpath(source_paths.paths)
    common_dir = commonpath
    if len(source_paths.paths) == 1:
        # If there is only one path, and it's a file, then commonpath will be that file, whereas
        # we want its enclosing dir. So detect that case.
        meta = await path_metadata_request(PathMetadataRequest(commonpath))
        # Chase any symlinks back to the final path they point to.
        while (
            meta.metadata
            and meta.metadata.kind == PathMetadataKind.SYMLINK
            and meta.metadata.symlink_target
        ):
            # NB: We don't `normpath` because eliminating `..` might change the meaning of the path
            #  if any of the intermediate directories are themselves symlinks.
            symlink_target = os.path.join(os.path.dirname(commonpath), meta.metadata.symlink_target)
            meta = await path_metadata_request(PathMetadataRequest(symlink_target))
        if meta.metadata and meta.metadata.kind == PathMetadataKind.FILE:
            common_dir = os.path.dirname(commonpath)
        else:
            common_dir = commonpath
    return CommonDir(Path(common_dir))


@dataclass(frozen=True)
class SourceDigest:
    digest: Digest


@rule
async def source_paths_to_digest(source_paths: SourcePaths) -> SourceDigest:
    source_digest = await path_globs_to_digest(
        PathGlobs(
            source_paths.path_strs(),
            glob_match_error_behavior=GlobMatchErrorBehavior.error,
            conjunction=GlobExpansionConjunction.all_match,
            description_of_origin="Input source paths",
        )
    )
    return SourceDigest(source_digest)


@dataclass(frozen=True)
class SourcePartition:
    """Access to source files and the config that goes with them."""

    _native_partition: PyNgSourcePartition
    _source_root: SourceRoot

    @memoized_property
    def source_paths(self) -> SourcePaths:
        return SourcePaths(
            tuple(Path(p) for p in self._native_partition.paths()), self._source_root
        )

    @memoized_property
    def options_reader(self) -> PyNgOptionsReader:
        return self._native_partition.options_reader()


class SourcePartitions(Collection[SourcePartition]):
    pass


# Uncacheable because we must get the most recent session value on each run.
@_uncacheable_rule
async def get_ng_options(session_values: SessionValues) -> PyNgOptions:
    return session_values[PyNgOptions]


# Uncacheable because we must recompute on each run.
@_uncacheable_rule
async def partition_sources(path_globs: PathGlobs) -> SourcePartitions:
    options = await get_ng_options(**implicitly())
    paths = await path_globs_to_paths(path_globs)
    # First partition by source root.
    source_roots = await get_source_roots(SourceRootsRequest.for_files(paths.files))
    root_to_paths = source_roots.root_to_paths()
    partitions: list[SourcePartition] = []
    for source_root, paths_in_partition in root_to_paths.items():
        # Then subpartition each of those by config.
        partitions.extend(
            SourcePartition(native_part, source_root)
            for native_part in options.partition_sources(
                tuple(str(path) for path in paths_in_partition)
            )
        )
    return SourcePartitions(tuple(partitions))


@rule
async def get_source_paths(partition: SourcePartition) -> SourcePaths:
    return partition.source_paths


def rules() -> tuple[Rule, ...]:
    return (*collect_rules(),)

1	# Copyright 2025 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	3✔
5
6	import os	3✔
7	from dataclasses import dataclass	3✔
8	from pathlib import Path	3✔
9
10	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior	3✔
11	from pants.engine.collection import Collection	3✔
12	from pants.engine.fs import GlobExpansionConjunction, PathGlobs, PathMetadataRequest	3✔
13	from pants.engine.internals.native_engine import (	3✔
14	Digest,
15	PathMetadataKind,
16	PyNgOptions,
17	PyNgOptionsReader,
18	PyNgSourcePartition,
19	)
20	from pants.engine.internals.session import SessionValues	3✔
21	from pants.engine.intrinsics import path_globs_to_digest, path_globs_to_paths, path_metadata_request	3✔
22	from pants.engine.rules import Rule, _uncacheable_rule, collect_rules, implicitly, rule	3✔
23	from pants.source.source_root import SourceRoot, SourceRootsRequest, get_source_roots	3✔
24	from pants.util.memo import memoized_property	3✔
25
26
27	@dataclass(frozen=True)	3✔
28	class SourcePaths:	3✔
29	"""A set of sources, under a single source root."""
30
31	paths: tuple[Path, ...]	3✔
32	source_root: SourceRoot	3✔
33
34	def path_strs(self) -> tuple[str, ...]:	3✔
35	return tuple(str(path) for path in self.paths)	×
36
37	def filter_by_suffixes(self, suffixes: tuple[str, ...]) -> SourcePaths:	3✔
UNCOV 38	suffixes_set = set(suffixes)	×
UNCOV 39	return SourcePaths(	×
40	tuple(path for path in self.paths if path.suffix in suffixes_set),
41	self.source_root,
42	)
43
44
45	@dataclass(frozen=True)	3✔
46	class CommonDir:	3✔
47	path: Path \| None	3✔
48
49
50	@rule	3✔
51	async def find_common_dir(source_paths: SourcePaths) -> CommonDir:	3✔
UNCOV 52	if not source_paths.paths: # We don't expect empty SourcePaths, but might as well be robust.	×
53	return CommonDir(None)	×
UNCOV 54	commonpath = os.path.commonpath(source_paths.paths)	×
UNCOV 55	common_dir = commonpath	×
UNCOV 56	if len(source_paths.paths) == 1:	×
57	# If there is only one path, and it's a file, then commonpath will be that file, whereas
58	# we want its enclosing dir. So detect that case.
UNCOV 59	meta = await path_metadata_request(PathMetadataRequest(commonpath))	×
60	# Chase any symlinks back to the final path they point to.
UNCOV 61	while (	×
62	meta.metadata
63	and meta.metadata.kind == PathMetadataKind.SYMLINK
64	and meta.metadata.symlink_target
65	):
66	# NB: We don't `normpath` because eliminating `..` might change the meaning of the path
67	# if any of the intermediate directories are themselves symlinks.
UNCOV 68	symlink_target = os.path.join(os.path.dirname(commonpath), meta.metadata.symlink_target)	×
UNCOV 69	meta = await path_metadata_request(PathMetadataRequest(symlink_target))	×
UNCOV 70	if meta.metadata and meta.metadata.kind == PathMetadataKind.FILE:	×
UNCOV 71	common_dir = os.path.dirname(commonpath)	×
72	else:
UNCOV 73	common_dir = commonpath	×
UNCOV 74	return CommonDir(Path(common_dir))	×
75
76
77	@dataclass(frozen=True)	3✔
78	class SourceDigest:	3✔
79	digest: Digest	3✔
80
81
82	@rule	3✔
83	async def source_paths_to_digest(source_paths: SourcePaths) -> SourceDigest:	3✔
84	source_digest = await path_globs_to_digest(	×
85	PathGlobs(
86	source_paths.path_strs(),
87	glob_match_error_behavior=GlobMatchErrorBehavior.error,
88	conjunction=GlobExpansionConjunction.all_match,
89	description_of_origin="Input source paths",
90	)
91	)
92	return SourceDigest(source_digest)	×
93
94
95	@dataclass(frozen=True)	3✔
96	class SourcePartition:	3✔
97	"""Access to source files and the config that goes with them."""
98
99	_native_partition: PyNgSourcePartition	3✔
100	_source_root: SourceRoot	3✔
101
102	@memoized_property	3✔
103	def source_paths(self) -> SourcePaths:	3✔
UNCOV 104	return SourcePaths(	×
105	tuple(Path(p) for p in self._native_partition.paths()), self._source_root
106	)
107
108	@memoized_property	3✔
109	def options_reader(self) -> PyNgOptionsReader:	3✔
110	return self._native_partition.options_reader()	×
111
112
113	class SourcePartitions(Collection[SourcePartition]):	3✔
114	pass	3✔
115
116
117	# Uncacheable because we must get the most recent session value on each run.
118	@_uncacheable_rule	3✔
119	async def get_ng_options(session_values: SessionValues) -> PyNgOptions:	3✔
120	return session_values[PyNgOptions]	×
121
122
123	# Uncacheable because we must recompute on each run.
124	@_uncacheable_rule	3✔
125	async def partition_sources(path_globs: PathGlobs) -> SourcePartitions:	3✔
UNCOV 126	options = await get_ng_options(**implicitly())	×
UNCOV 127	paths = await path_globs_to_paths(path_globs)	×
128	# First partition by source root.
UNCOV 129	source_roots = await get_source_roots(SourceRootsRequest.for_files(paths.files))	×
UNCOV 130	root_to_paths = source_roots.root_to_paths()	×
UNCOV 131	partitions: list[SourcePartition] = []	×
UNCOV 132	for source_root, paths_in_partition in root_to_paths.items():	×
133	# Then subpartition each of those by config.
UNCOV 134	partitions.extend(	×
135	SourcePartition(native_part, source_root)
136	for native_part in options.partition_sources(
137	tuple(str(path) for path in paths_in_partition)
138	)
139	)
UNCOV 140	return SourcePartitions(tuple(partitions))	×
141
142
143	@rule	3✔
144	async def get_source_paths(partition: SourcePartition) -> SourcePaths:	3✔
145	return partition.source_paths	×
146
147
148	def rules() -> tuple[Rule, ...]:	3✔
149	return (*collect_rules(),)	3✔

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