26080722777

Committed 19 May 2026 06:37AM UTC coverage: 52.106% (-11.5%) from 63.597%

Build # 26080722777

Build Type

Pull #23250

github

Committed by

web-flow

Commit Message

Merge 63ec06323 into 2693df832

Pull Request Pull Request #23250: Feature: Add generic option to docker image

Coverage Stats

12 of 50 new or added lines in 3 files covered. (24.0%)

5382 existing lines in 201 files now uncovered.

32053 of 61515 relevant lines covered (52.11%)

1.04 hits per line

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

59.74

/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	2✔
5
6	import os	2✔
7	from dataclasses import dataclass	2✔
8	from pathlib import Path	2✔
9
10	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior	2✔
11	from pants.engine.collection import Collection	2✔
12	from pants.engine.fs import GlobExpansionConjunction, PathGlobs, PathMetadataRequest	2✔
13	from pants.engine.internals.native_engine import (	2✔
14	Digest,
15	PathMetadataKind,
16	PyNgOptions,
17	PyNgOptionsReader,
18	PyNgSourcePartition,
19	)
20	from pants.engine.internals.session import SessionValues	2✔
21	from pants.engine.intrinsics import path_globs_to_digest, path_globs_to_paths, path_metadata_request	2✔
22	from pants.engine.rules import Rule, _uncacheable_rule, collect_rules, implicitly, rule	2✔
23	from pants.source.source_root import SourceRoot, SourceRootsRequest, get_source_roots	2✔
24	from pants.util.memo import memoized_property	2✔
25
26
27	@dataclass(frozen=True)	2✔
28	class SourcePaths:	2✔
29	"""A set of sources, under a single source root."""
30
31	paths: tuple[Path, ...]	2✔
32	source_root: SourceRoot	2✔
33
34	def path_strs(self) -> tuple[str, ...]:	2✔
35	return tuple(str(path) for path in self.paths)	×
36
37	def filter_by_suffixes(self, suffixes: tuple[str, ...]) -> SourcePaths:	2✔
38	suffixes_set = set(suffixes)	×
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)	2✔
46	class CommonDir:	2✔
47	path: Path \| None	2✔
48
49
50	@rule	2✔
51	async def find_common_dir(source_paths: SourcePaths) -> CommonDir:	2✔
52	if not source_paths.paths: # We don't expect empty SourcePaths, but might as well be robust.	×
53	return CommonDir(None)	×
54	commonpath = os.path.commonpath(source_paths.paths)	×
55	common_dir = commonpath	×
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.
59	meta = await path_metadata_request(PathMetadataRequest(commonpath))	×
60	# Chase any symlinks back to the final path they point to.
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.
68	symlink_target = os.path.join(os.path.dirname(commonpath), meta.metadata.symlink_target)	×
69	meta = await path_metadata_request(PathMetadataRequest(symlink_target))	×
70	if meta.metadata and meta.metadata.kind == PathMetadataKind.FILE:	×
71	common_dir = os.path.dirname(commonpath)	×
72	else:
73	common_dir = commonpath	×
74	return CommonDir(Path(common_dir))	×
75
76
77	@dataclass(frozen=True)	2✔
78	class SourceDigest:	2✔
79	digest: Digest	2✔
80
81
82	@rule	2✔
83	async def source_paths_to_digest(source_paths: SourcePaths) -> SourceDigest:	2✔
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)	2✔
96	class SourcePartition:	2✔
97	"""Access to source files and the config that goes with them."""
98
99	_native_partition: PyNgSourcePartition	2✔
100	_source_root: SourceRoot	2✔
101
102	@memoized_property	2✔
103	def source_paths(self) -> SourcePaths:	2✔
104	return SourcePaths(	×
105	tuple(Path(p) for p in self._native_partition.paths()), self._source_root
106	)
107
108	@memoized_property	2✔
109	def options_reader(self) -> PyNgOptionsReader:	2✔
110	return self._native_partition.options_reader()	×
111
112
113	class SourcePartitions(Collection[SourcePartition]):	2✔
114	pass	2✔
115
116
117	# Uncacheable because we must get the most recent session value on each run.
118	@_uncacheable_rule	2✔
119	async def get_ng_options(session_values: SessionValues) -> PyNgOptions:	2✔
120	return session_values[PyNgOptions]	×
121
122
123	# Uncacheable because we must recompute on each run.
124	@_uncacheable_rule	2✔
125	async def partition_sources(path_globs: PathGlobs) -> SourcePartitions:	2✔
126	options = await get_ng_options(**implicitly())	×
127	paths = await path_globs_to_paths(path_globs)	×
128	# First partition by source root.
129	source_roots = await get_source_roots(SourceRootsRequest.for_files(paths.files))	×
130	root_to_paths = source_roots.root_to_paths()	×
131	partitions: list[SourcePartition] = []	×
132	for source_root, paths_in_partition in root_to_paths.items():	×
133	# Then subpartition each of those by config.
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	)
140	return SourcePartitions(tuple(partitions))	×
141
142
143	@rule	2✔
144	async def get_source_paths(partition: SourcePartition) -> SourcePaths:	2✔
145	return partition.source_paths	×
146
147
148	def rules() -> tuple[Rule, ...]:	2✔
UNCOV 149	return (*collect_rules(),)	×

pantsbuild / pants / 26080722777

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