24055979590

Committed 06 Apr 2026 11:17PM UTC coverage: 52.37% (-40.5%) from 92.908%

Build # 24055979590

Build Type

Pull #23225

github

Committed by

web-flow

Commit Message

Merge 67474653c into 542ca048d

Pull Request Pull Request #23225: Add --test-show-all-batch-targets to expose all targets in batched pytest

Coverage Stats

6 of 17 new or added lines in 2 files covered. (35.29%)

23030 existing lines in 605 files now uncovered.

31643 of 60422 relevant lines covered (52.37%)

1.05 hits per line

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

0.0

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

1	# Copyright 2025 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
UNCOV 4	from __future__ import annotations	×
5
UNCOV 6	import os	×
UNCOV 7	from dataclasses import dataclass	×
UNCOV 8	from pathlib import Path	×
9
UNCOV 10	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior	×
UNCOV 11	from pants.engine.collection import Collection	×
UNCOV 12	from pants.engine.fs import GlobExpansionConjunction, PathGlobs, PathMetadataRequest	×
UNCOV 13	from pants.engine.internals.native_engine import (	×
14	Digest,
15	PathMetadataKind,
16	PyNgOptions,
17	PyNgOptionsReader,
18	PyNgSourcePartition,
19	)
UNCOV 20	from pants.engine.internals.session import SessionValues	×
UNCOV 21	from pants.engine.intrinsics import path_globs_to_digest, path_globs_to_paths, path_metadata_request	×
UNCOV 22	from pants.engine.rules import Rule, _uncacheable_rule, collect_rules, implicitly, rule	×
UNCOV 23	from pants.source.source_root import SourceRoot, SourceRootsRequest, get_source_roots	×
UNCOV 24	from pants.util.memo import memoized_property	×
25
26
UNCOV 27	@dataclass(frozen=True)	×
UNCOV 28	class SourcePaths:	×
29	"""A set of sources, under a single source root."""
30
UNCOV 31	paths: tuple[Path, ...]	×
UNCOV 32	source_root: SourceRoot	×
33
UNCOV 34	def path_strs(self) -> tuple[str, ...]:	×
35	return tuple(str(path) for path in self.paths)	×
36
UNCOV 37	def filter_by_suffixes(self, suffixes: tuple[str, ...]) -> SourcePaths:	×
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
UNCOV 45	@dataclass(frozen=True)	×
UNCOV 46	class CommonDir:	×
UNCOV 47	path: Path \| None	×
48
49
UNCOV 50	@rule	×
UNCOV 51	async def find_common_dir(source_paths: SourcePaths) -> CommonDir:	×
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
UNCOV 77	@dataclass(frozen=True)	×
UNCOV 78	class SourceDigest:	×
UNCOV 79	digest: Digest	×
80
81
UNCOV 82	@rule	×
UNCOV 83	async def source_paths_to_digest(source_paths: SourcePaths) -> SourceDigest:	×
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
UNCOV 95	@dataclass(frozen=True)	×
UNCOV 96	class SourcePartition:	×
97	"""Access to source files and the config that goes with them."""
98
UNCOV 99	_native_partition: PyNgSourcePartition	×
UNCOV 100	_source_root: SourceRoot	×
101
UNCOV 102	@memoized_property	×
UNCOV 103	def source_paths(self) -> SourcePaths:	×
UNCOV 104	return SourcePaths(	×
105	tuple(Path(p) for p in self._native_partition.paths()), self._source_root
106	)
107
UNCOV 108	@memoized_property	×
UNCOV 109	def options_reader(self) -> PyNgOptionsReader:	×
110	return self._native_partition.options_reader()	×
111
112
UNCOV 113	class SourcePartitions(Collection[SourcePartition]):	×
UNCOV 114	pass	×
115
116
117	# Uncacheable because we must get the most recent session value on each run.
UNCOV 118	@_uncacheable_rule	×
UNCOV 119	async def get_ng_options(session_values: SessionValues) -> PyNgOptions:	×
120	return session_values[PyNgOptions]	×
121
122
123	# Uncacheable because we must recompute on each run.
UNCOV 124	@_uncacheable_rule	×
UNCOV 125	async def partition_sources(path_globs: PathGlobs) -> SourcePartitions:	×
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
UNCOV 143	@rule	×
UNCOV 144	async def get_source_paths(partition: SourcePartition) -> SourcePaths:	×
145	return partition.source_paths	×
146
147
UNCOV 148	def rules() -> tuple[Rule, ...]:	×
UNCOV 149	return (*collect_rules(),)	×

pantsbuild / pants / 24055979590

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