19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

0.0

/src/python/pants/backend/python/util_rules/entry_points.py

1	# Copyright 2024 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
UNCOV 3	from collections import defaultdict	×
UNCOV 4	from collections.abc import Callable, Iterable	×
UNCOV 5	from dataclasses import dataclass	×
6
UNCOV 7	from pants.backend.python.dependency_inference.module_mapper import (	×
8	PythonModuleOwnersRequest,
9	map_module_to_address,
10	)
UNCOV 11	from pants.backend.python.goals.pytest_runner import PytestPluginSetup, PytestPluginSetupRequest	×
UNCOV 12	from pants.backend.python.target_types import (	×
13	EntryPoint,
14	PythonDistribution,
15	PythonDistributionDependenciesField,
16	PythonDistributionEntryPoint,
17	PythonDistributionEntryPointsField,
18	PythonTestsDependenciesField,
19	PythonTestsEntryPointDependenciesField,
20	PythonTestsGeneratorTarget,
21	PythonTestTarget,
22	ResolvedPythonDistributionEntryPoints,
23	ResolvePythonDistributionEntryPointsRequest,
24	)
UNCOV 25	from pants.backend.python.target_types_rules import (	×
26	get_python_distribution_entry_point_unambiguous_module_owners,
27	resolve_python_distribution_entry_points,
28	)
UNCOV 29	from pants.engine.addresses import Addresses, UnparsedAddressInputs	×
UNCOV 30	from pants.engine.fs import CreateDigest, FileContent, PathGlobs, Paths	×
UNCOV 31	from pants.engine.internals.graph import determine_explicitly_provided_dependencies, resolve_targets	×
UNCOV 32	from pants.engine.internals.native_engine import EMPTY_DIGEST, Address, Digest	×
UNCOV 33	from pants.engine.internals.selectors import concurrently	×
UNCOV 34	from pants.engine.intrinsics import create_digest, path_globs_to_paths	×
UNCOV 35	from pants.engine.rules import collect_rules, implicitly, rule	×
UNCOV 36	from pants.engine.target import (	×
37	DependenciesRequest,
38	ExplicitlyProvidedDependencies,
39	FieldSet,
40	InferDependenciesRequest,
41	InferredDependencies,
42	Target,
43	Targets,
44	)
UNCOV 45	from pants.engine.unions import UnionRule	×
UNCOV 46	from pants.util.frozendict import FrozenDict	×
UNCOV 47	from pants.util.logging import LogLevel	×
UNCOV 48	from pants.util.ordered_set import FrozenOrderedSet, OrderedSet	×
49
UNCOV 50	PythonDistributionEntryPointGroupPredicate = Callable[[Target, str], bool]	×
UNCOV 51	PythonDistributionEntryPointPredicate = Callable[[Target, str, str], bool]	×
52
53
UNCOV 54	@dataclass(frozen=True)	×
UNCOV 55	class GetEntryPointDependenciesRequest:	×
56	targets: Targets
57	group_predicate: PythonDistributionEntryPointGroupPredicate
58	predicate: PythonDistributionEntryPointPredicate
59
60
UNCOV 61	@dataclass(frozen=True)	×
UNCOV 62	class EntryPointDependencies:	×
63	addresses: FrozenOrderedSet[Address]
64
UNCOV 65	def __init__(self, addresses: Iterable[Address]) -> None:	×
66	object.__setattr__(self, "addresses", FrozenOrderedSet(sorted(addresses)))	×
67
68
UNCOV 69	@rule	×
UNCOV 70	async def get_filtered_entry_point_dependencies(	×
71	request: GetEntryPointDependenciesRequest,
72	) -> EntryPointDependencies:
73	# This is based on pants.backend.python.target_type_rules.infer_python_distribution_dependencies,
74	# but handles multiple targets and filters the entry_points to just get the requested deps.
75	all_explicit_dependencies = await concurrently(	×
76	determine_explicitly_provided_dependencies(
77	**implicitly(DependenciesRequest(tgt[PythonDistributionDependenciesField]))
78	)
79	for tgt in request.targets
80	)
81	# TODO: This is a circular import on call-by-name
82	resolved_entry_points = await concurrently(	×
83	resolve_python_distribution_entry_points(
84	ResolvePythonDistributionEntryPointsRequest(tgt[PythonDistributionEntryPointsField])
85	)
86	for tgt in request.targets
87	)
88
89	filtered_entry_point_pex_addresses: list[Address] = []	×
90	filtered_entry_point_modules: list[	×
91	tuple[Address, str, str, EntryPoint, ExplicitlyProvidedDependencies]
92	] = []
93
94	distribution_entry_points: ResolvedPythonDistributionEntryPoints
95	for tgt, distribution_entry_points, explicitly_provided_deps in zip(	×
96	request.targets, resolved_entry_points, all_explicit_dependencies
97	):
98	# use .val instead of .explicit_modules and .pex_binary_addresses to facilitate filtering
99	for ep_group, entry_points in distribution_entry_points.val.items():	×
100	if not request.group_predicate(tgt, ep_group):	×
101	continue	×
102	for ep_name, ep_val in entry_points.items():	×
103	if not request.predicate(tgt, ep_group, ep_name):	×
104	continue	×
105	if ep_val.pex_binary_address:	×
106	filtered_entry_point_pex_addresses.append(ep_val.pex_binary_address)	×
107	else:
108	filtered_entry_point_modules.append(	×
109	(
110	tgt.address,
111	ep_group,
112	ep_name,
113	ep_val.entry_point,
114	explicitly_provided_deps,
115	)
116	)
117	filtered_module_owners = await concurrently(	×
118	map_module_to_address(
119	PythonModuleOwnersRequest(entry_point.module, resolve=None), **implicitly()
120	)
121	for _, _, _, entry_point, _ in filtered_entry_point_modules
122	)
123
124	filtered_unambiguous_module_owners: OrderedSet[Address] = OrderedSet()	×
125	for (address, ep_group, ep_name, entry_point, explicitly_provided_deps), owners in zip(	×
126	filtered_entry_point_modules, filtered_module_owners
127	):
128	filtered_unambiguous_module_owners.update(	×
129	get_python_distribution_entry_point_unambiguous_module_owners(
130	address, ep_group, ep_name, entry_point, explicitly_provided_deps, owners
131	)
132	)
133
134	return EntryPointDependencies(	×
135	Addresses((filtered_entry_point_pex_addresses, filtered_unambiguous_module_owners))
136	)
137
138
UNCOV 139	async def _get_entry_point_deps_targets_and_predicates(	×
140	owning_address: Address, entry_point_deps: PythonTestsEntryPointDependenciesField
141	) -> tuple[
142	Targets, PythonDistributionEntryPointGroupPredicate, PythonDistributionEntryPointPredicate
143	]:
144	if not entry_point_deps.value: # type narrowing for mypy	×
145	raise ValueError(	×
146	"Please file an issue if you see this error. This rule helper must not "
147	"be called with an empty entry_point_dependencies field."
148	)
149	targets = await resolve_targets(	×
150	**implicitly(
151	UnparsedAddressInputs(
152	entry_point_deps.value.keys(),
153	owning_address=owning_address,
154	description_of_origin=f"{PythonTestsEntryPointDependenciesField.alias} from {owning_address}",
155	)
156	)
157	)
158
159	requested_entry_points: dict[Target, set[str]] = {}	×
160
161	requested_ep: tuple[str, ...]
162	for target, requested_ep in zip(targets, entry_point_deps.value.values()):	×
163	if not requested_ep:	×
164	# requested an empty list, so no entry points were actually requested.
165	continue	×
166	if "*" in requested_ep and len(requested_ep) > 1:	×
167	requested_ep = ("*",)	×
168
169	if not target.has_field(PythonDistributionEntryPointsField):	×
170	# unknown target type. ignore
171	continue	×
172	if not target.get(PythonDistributionEntryPointsField).value:	×
173	# no entry points can be resolved.
174	# TODO: Maybe warn that the requested entry points do not exist?
175	continue	×
176	requested_entry_points[target] = set(requested_ep)	×
177
178	def group_predicate(tgt: Target, ep_group: str) -> bool:	×
179	relevant = ("*", ep_group)	×
180	for item in sorted(requested_entry_points[tgt]):	×
181	if item in relevant or item.startswith(f"{ep_group}/"):	×
182	return True	×
183	return False	×
184
185	def predicate(tgt: Target, ep_group: str, ep_name: str) -> bool:	×
186	relevant = {"*", ep_group, f"{ep_group}/{ep_name}"}	×
187	if relevant & requested_entry_points[tgt]:	×
188	# at least one requested entry point is relevant
189	return True	×
190	return False	×
191
192	return Targets(requested_entry_points.keys()), group_predicate, predicate	×
193
194
UNCOV 195	@dataclass(frozen=True)	×
UNCOV 196	class PythonTestsEntryPointDependenciesInferenceFieldSet(FieldSet):	×
UNCOV 197	required_fields = (	×
198	PythonTestsDependenciesField,
199	PythonTestsEntryPointDependenciesField,
200	)
201	entry_point_dependencies: PythonTestsEntryPointDependenciesField
202
203
UNCOV 204	class InferEntryPointDependencies(InferDependenciesRequest):	×
UNCOV 205	infer_from = PythonTestsEntryPointDependenciesInferenceFieldSet	×
206
207
UNCOV 208	@rule(	×
209	desc=f"Infer dependencies based on `{PythonTestsEntryPointDependenciesField.alias}` field.",
210	level=LogLevel.DEBUG,
211	)
UNCOV 212	async def infer_entry_point_dependencies(	×
213	request: InferEntryPointDependencies,
214	) -> InferredDependencies:
215	entry_point_deps: PythonTestsEntryPointDependenciesField = (	×
216	request.field_set.entry_point_dependencies
217	)
218	if entry_point_deps.value is None:	×
219	return InferredDependencies([])	×
220
221	dist_targets, group_predicate, predicate = await _get_entry_point_deps_targets_and_predicates(	×
222	request.field_set.address, entry_point_deps
223	)
224
225	entry_point_dependencies = await get_filtered_entry_point_dependencies(	×
226	GetEntryPointDependenciesRequest(dist_targets, group_predicate, predicate)
227	)
228	return InferredDependencies(entry_point_dependencies.addresses)	×
229
230
UNCOV 231	@dataclass(frozen=True)	×
UNCOV 232	class GenerateEntryPointsTxtRequest:	×
233	targets: Targets
234	group_predicate: PythonDistributionEntryPointGroupPredicate
235	predicate: PythonDistributionEntryPointPredicate
236
237
UNCOV 238	@dataclass(frozen=True)	×
UNCOV 239	class EntryPointsTxt:	×
240	digest: Digest
241
242
UNCOV 243	@rule	×
UNCOV 244	async def generate_entry_points_txt(request: GenerateEntryPointsTxtRequest) -> EntryPointsTxt:	×
245	if not request.targets:	×
246	return EntryPointsTxt(EMPTY_DIGEST)	×
247
248	all_resolved_entry_points = await concurrently(	×
249	resolve_python_distribution_entry_points(
250	ResolvePythonDistributionEntryPointsRequest(tgt[PythonDistributionEntryPointsField])
251	)
252	for tgt in request.targets
253	)
254
255	possible_paths = [	×
256	{
257	f"{tgt.address.spec_path}/{ep.entry_point.module.split('.')[0]}"
258	for _, entry_points in (resolved_eps.val or {}).items()
259	for ep in entry_points.values()
260	}
261	for tgt, resolved_eps in zip(request.targets, all_resolved_entry_points)
262	]
263	resolved_paths = await concurrently(	×
264	path_globs_to_paths(PathGlobs(module_candidate_paths))
265	for module_candidate_paths in possible_paths
266	)
267
268	entry_points_by_path: dict[str, list[tuple[Target, ResolvedPythonDistributionEntryPoints]]] = (	×
269	defaultdict(list)
270	)
271
272	target: Target
273	resolved_ep: ResolvedPythonDistributionEntryPoints
274	paths: Paths
275	for target, resolved_ep, paths in zip(	×
276	request.targets, all_resolved_entry_points, resolved_paths
277	):
278	path = paths.dirs[0] # just take the first match	×
279	entry_points_by_path[path].append((target, resolved_ep))	×
280
281	entry_points_txt_files = []	×
282	for module_path, target_and_resolved_eps in entry_points_by_path.items():	×
283	group_sections = {}	×
284
285	for target, resolved_ep in target_and_resolved_eps:	×
286	ep_group: str
287	entry_points: FrozenDict[str, PythonDistributionEntryPoint]
288	for ep_group, entry_points in resolved_ep.val.items():	×
289	if not entry_points or not request.group_predicate(target, ep_group):	×
290	continue	×
291
292	entry_points_txt_section = f"[{ep_group}]\n"	×
293	selected_entry_points_in_group = False	×
294	for entry_point_name, ep in sorted(entry_points.items()):	×
295	if not request.predicate(target, ep_group, entry_point_name):	×
296	continue	×
297	selected_entry_points_in_group = True	×
298	entry_points_txt_section += f"{entry_point_name} = {ep.entry_point.spec}\n"	×
299	if not selected_entry_points_in_group:	×
300	continue	×
301	entry_points_txt_section += "\n"	×
302	group_sections[ep_group] = entry_points_txt_section	×
303
304	if not group_sections:	×
305	continue	×
306
307	# consistent sorting
308	entry_points_txt_contents = "".join(	×
309	group_sections[ep_group] for ep_group in sorted(group_sections)
310	)
311
312	entry_points_txt_path = f"{module_path}.egg-info/entry_points.txt"	×
313	entry_points_txt_files.append(	×
314	FileContent(entry_points_txt_path, entry_points_txt_contents.encode("utf-8"))
315	)
316
317	if not entry_points_txt_files:	×
318	digest = EMPTY_DIGEST	×
319	else:
320	digest = await create_digest(CreateDigest(entry_points_txt_files))	×
321	return EntryPointsTxt(digest)	×
322
323
UNCOV 324	class GenerateEntryPointsTxtFromEntryPointDependenciesRequest(PytestPluginSetupRequest):	×
UNCOV 325	@classmethod	×
UNCOV 326	def is_applicable(cls, target: Target) -> bool:	×
327	# select python_tests targets with entry_point_dependencies field
328	return (	×
329	target.has_field(PythonTestsEntryPointDependenciesField)
330	and target.get(PythonTestsEntryPointDependenciesField).value is not None
331	)
332
333
UNCOV 334	@rule(	×
335	desc=f"Generate entry_points.txt to imitate `{PythonDistribution.alias}` installation.",
336	level=LogLevel.DEBUG,
337	)
UNCOV 338	async def generate_entry_points_txt_from_entry_point_dependencies(	×
339	request: GenerateEntryPointsTxtFromEntryPointDependenciesRequest,
340	) -> PytestPluginSetup:
341	entry_point_deps = request.target[PythonTestsEntryPointDependenciesField]	×
342	if not entry_point_deps.value:	×
343	return PytestPluginSetup(EMPTY_DIGEST)	×
344
345	dist_targets, group_predicate, predicate = await _get_entry_point_deps_targets_and_predicates(	×
346	request.target.address, entry_point_deps
347	)
348
349	entry_points_txt = await generate_entry_points_txt(	×
350	GenerateEntryPointsTxtRequest(dist_targets, group_predicate, predicate)
351	)
352	return PytestPluginSetup(entry_points_txt.digest)	×
353
354
UNCOV 355	def rules():	×
UNCOV 356	return [	×
357	*collect_rules(),
358	# TODO: remove these register_plugin_field calls once this moves out of experimental
359	PythonTestTarget.register_plugin_field(PythonTestsEntryPointDependenciesField),
360	PythonTestsGeneratorTarget.register_plugin_field(
361	PythonTestsEntryPointDependenciesField,
362	as_moved_field=True,
363	),
364	UnionRule(InferDependenciesRequest, InferEntryPointDependencies),
365	UnionRule(
366	PytestPluginSetupRequest,
367	GenerateEntryPointsTxtFromEntryPointDependenciesRequest,
368	),
369	]

pantsbuild / pants / 19015773527

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