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/helm/util_rules/sources.py

1	# Copyright 2022 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	×
8
UNCOV 9	from pants.backend.helm.target_types import (	×
10	HelmChartFieldSet,
11	HelmChartMetaSourceField,
12	HelmChartSourcesField,
13	)
UNCOV 14	from pants.core.target_types import FileSourceField, ResourceSourceField	×
UNCOV 15	from pants.core.util_rules import source_files	×
UNCOV 16	from pants.core.util_rules.source_files import (	×
17	SourceFiles,
18	SourceFilesRequest,
19	determine_source_files,
20	)
UNCOV 21	from pants.engine.engine_aware import EngineAwareParameter	×
UNCOV 22	from pants.engine.fs import DigestSubset, MergeDigests, PathGlobs, Snapshot	×
UNCOV 23	from pants.engine.internals.graph import hydrate_sources, resolve_targets	×
UNCOV 24	from pants.engine.internals.native_engine import RemovePrefix	×
UNCOV 25	from pants.engine.intrinsics import digest_subset_to_digest, digest_to_snapshot	×
UNCOV 26	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	×
UNCOV 27	from pants.engine.target import DependenciesRequest, HydrateSourcesRequest, SourcesField, Target	×
28
29
UNCOV 30	@dataclass(frozen=True)	×
UNCOV 31	class HelmChartRootRequest(EngineAwareParameter):	×
UNCOV 32	source: HelmChartMetaSourceField	×
33
UNCOV 34	def debug_hint(self) -> str \| None:	×
35	return self.source.address.spec	×
36
37
UNCOV 38	@dataclass(frozen=True)	×
UNCOV 39	class HelmChartRoot:	×
UNCOV 40	path: str	×
41
42
UNCOV 43	@rule(desc="Detect Helm chart source root")	×
UNCOV 44	async def find_chart_source_root(request: HelmChartRootRequest) -> HelmChartRoot:	×
45	source = await hydrate_sources(	×
46	HydrateSourcesRequest(
47	request.source, for_sources_types=[HelmChartMetaSourceField], enable_codegen=True
48	),
49	**implicitly(),
50	)
51	assert len(source.snapshot.files) == 1	×
52
53	return HelmChartRoot(os.path.dirname(source.snapshot.files[0]))	×
54
55
UNCOV 56	@dataclass(frozen=True)	×
UNCOV 57	class HelmChartSourceFilesRequest(EngineAwareParameter):	×
UNCOV 58	field_set: HelmChartFieldSet	×
UNCOV 59	include_resources: bool	×
UNCOV 60	include_files: bool	×
UNCOV 61	include_metadata: bool	×
62
UNCOV 63	@classmethod	×
UNCOV 64	def create(	×
65	cls,
66	target: Target,
67	*,
68	include_resources: bool = True,
69	include_files: bool = False,
70	include_metadata: bool = True,
71	) -> HelmChartSourceFilesRequest:
UNCOV 72	return cls.for_field_set(	×
73	HelmChartFieldSet.create(target),
74	include_resources=include_resources,
75	include_files=include_files,
76	include_metadata=include_metadata,
77	)
78
UNCOV 79	@classmethod	×
UNCOV 80	def for_field_set(	×
81	cls,
82	field_set: HelmChartFieldSet,
83	*,
84	include_resources: bool = True,
85	include_files: bool = False,
86	include_metadata: bool = True,
87	) -> HelmChartSourceFilesRequest:
UNCOV 88	return cls(	×
89	field_set=field_set,
90	include_resources=include_resources,
91	include_files=include_files,
92	include_metadata=include_metadata,
93	)
94
UNCOV 95	@property	×
UNCOV 96	def sources_fields(self) -> tuple[SourcesField, ...]:	×
97	fields: list[SourcesField] = [self.field_set.sources]	×
98	if self.include_metadata:	×
99	fields.append(self.field_set.chart)	×
100	return tuple(fields)	×
101
UNCOV 102	@property	×
UNCOV 103	def valid_sources_types(self) -> tuple[type[SourcesField], ...]:	×
104	types: list[type[SourcesField]] = [HelmChartSourcesField]	×
105	if self.include_metadata:	×
106	types.append(HelmChartMetaSourceField)	×
107	if self.include_resources:	×
108	types.append(ResourceSourceField)	×
109	if self.include_files:	×
110	types.append(FileSourceField)	×
111	return tuple(types)	×
112
UNCOV 113	def debug_hint(self) -> str \| None:	×
114	return self.field_set.address.spec	×
115
116
UNCOV 117	@dataclass(frozen=True)	×
UNCOV 118	class HelmChartSourceFiles:	×
UNCOV 119	snapshot: Snapshot	×
UNCOV 120	unrooted_files: tuple[str, ...]	×
121
122
UNCOV 123	async def _strip_chart_source_root(	×
124	source_files: SourceFiles, chart_root: HelmChartRoot
125	) -> Snapshot:
126	if not source_files.snapshot.files:	×
127	return source_files.snapshot	×
128
129	if source_files.unrooted_files:	×
130	rooted_files = set(source_files.snapshot.files) - set(source_files.unrooted_files)	×
131	rooted_files_snapshot = await digest_to_snapshot(	×
132	**implicitly(DigestSubset(source_files.snapshot.digest, PathGlobs(rooted_files)))
133	)
134	else:
135	rooted_files_snapshot = source_files.snapshot	×
136
137	resulting_snapshot = await digest_to_snapshot(	×
138	**implicitly(RemovePrefix(rooted_files_snapshot.digest, chart_root.path))
139	)
140	if source_files.unrooted_files:	×
141	# Add unrooted files back in
142	unrooted_digest = await digest_subset_to_digest(	×
143	DigestSubset(source_files.snapshot.digest, PathGlobs(source_files.unrooted_files)),
144	)
145	resulting_snapshot = await digest_to_snapshot(	×
146	**implicitly(MergeDigests([resulting_snapshot.digest, unrooted_digest]))
147	)
148
149	return resulting_snapshot	×
150
151
UNCOV 152	@rule	×
UNCOV 153	async def get_helm_source_files(request: HelmChartSourceFilesRequest) -> HelmChartSourceFiles:	×
154	chart_root, dependencies = await concurrently(	×
155	find_chart_source_root(HelmChartRootRequest(request.field_set.chart)),
156	resolve_targets(**implicitly(DependenciesRequest(request.field_set.dependencies))),
157	)
158
159	source_files, original_sources = await concurrently(	×
160	determine_source_files(
161	SourceFilesRequest(
162	sources_fields=[
163	*request.sources_fields,
164	*(
165	tgt.get(SourcesField)
166	for tgt in dependencies
167	if not HelmChartFieldSet.is_applicable(tgt)
168	),
169	],
170	for_sources_types=request.valid_sources_types,
171	enable_codegen=True,
172	)
173	),
174	determine_source_files(
175	SourceFilesRequest([request.field_set.sources], enable_codegen=False)
176	),
177	)
178
179	stripped_source_files = await _strip_chart_source_root(source_files, chart_root)	×
180	stripped_original_sources = await _strip_chart_source_root(original_sources, chart_root)	×
181
182	all_files_snapshot = await digest_to_snapshot(	×
183	**implicitly(MergeDigests([stripped_source_files.digest, stripped_original_sources.digest]))
184	)
185	return HelmChartSourceFiles(	×
186	snapshot=all_files_snapshot,
187	unrooted_files=(source_files.unrooted_files, original_sources.unrooted_files),
188	)
189
190
UNCOV 191	def rules():	×
UNCOV 192	return [collect_rules(), source_files.rules()]	×

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