18913303678

Committed 29 Oct 2025 03:29PM UTC coverage: 80.004% (-0.3%) from 80.283%

Build # 18913303678

Build Type

push

github

Committed by

web-flow

Commit Message

Updating 3rd party lockfiles for PBS script and MyPy (#22796)

Also small bumps to fastapi and starlette, which updated over the last few days. Starlette updated for features, and then a security vulnerability.

Fastapi bumped just for starlette.

Run Details

88 of 93 new or added lines in 73 files covered. (94.62%)

221 existing lines in 15 files now uncovered.

77334 of 96663 relevant lines covered (80.0%)

5.2 hits per line

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

0.0

/src/python/pants/backend/helm/test/unittest.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 dataclasses	×
UNCOV 7	import logging	×
UNCOV 8	import os	×
UNCOV 9	from dataclasses import dataclass	×
UNCOV 10	from typing import Any	×
11
UNCOV 12	from pants.backend.helm.dependency_inference.unittest import rules as dependency_rules	×
UNCOV 13	from pants.backend.helm.subsystems.unittest import HelmUnitTestSubsystem	×
UNCOV 14	from pants.backend.helm.subsystems.unittest import rules as subsystem_rules	×
UNCOV 15	from pants.backend.helm.target_types import (	×
16	HelmChartFieldSet,
17	HelmChartMetaSourceField,
18	HelmChartTarget,
19	HelmUnitTestDependenciesField,
20	HelmUnitTestSourceField,
21	HelmUnitTestStrictField,
22	HelmUnitTestTestsGeneratorTarget,
23	HelmUnitTestTestTarget,
24	HelmUnitTestTimeoutField,
25	)
UNCOV 26	from pants.backend.helm.util_rules import tool	×
UNCOV 27	from pants.backend.helm.util_rules.chart import HelmChart, HelmChartRequest, get_helm_chart	×
UNCOV 28	from pants.backend.helm.util_rules.sources import (	×
29	HelmChartRoot,
30	HelmChartRootRequest,
31	find_chart_source_root,
32	)
UNCOV 33	from pants.backend.helm.util_rules.tool import HelmProcess, helm_process	×
UNCOV 34	from pants.core.goals.generate_snapshots import GenerateSnapshotsFieldSet, GenerateSnapshotsResult	×
UNCOV 35	from pants.core.goals.test import TestFieldSet, TestRequest, TestResult, TestSubsystem	×
UNCOV 36	from pants.core.target_types import FileSourceField, ResourceSourceField	×
UNCOV 37	from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files	×
UNCOV 38	from pants.core.util_rules.stripped_source_files import strip_source_roots	×
UNCOV 39	from pants.engine.addresses import Address	×
UNCOV 40	from pants.engine.fs import AddPrefix, DigestSubset, MergeDigests, PathGlobs, RemovePrefix	×
UNCOV 41	from pants.engine.internals.graph import resolve_targets	×
UNCOV 42	from pants.engine.internals.graph import transitive_targets as transitive_targets_get	×
UNCOV 43	from pants.engine.intrinsics import (	×
44	add_prefix,
45	digest_subset_to_digest,
46	digest_to_snapshot,
47	execute_process_with_retry,
48	merge_digests,
49	remove_prefix,
50	)
UNCOV 51	from pants.engine.process import (	×
52	ProcessCacheScope,
53	ProcessWithRetries,
54	fallible_to_exec_result_or_raise,
55	)
UNCOV 56	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	×
UNCOV 57	from pants.engine.target import DependenciesRequest, SourcesField, TransitiveTargetsRequest	×
UNCOV 58	from pants.engine.unions import UnionRule	×
UNCOV 59	from pants.util.logging import LogLevel	×
UNCOV 60	from pants.util.strutil import Simplifier	×
61
UNCOV 62	logger = logging.getLogger(__name__)	×
63
64
UNCOV 65	class MissingUnitTestChartDependency(Exception):	×
UNCOV 66	f"""Indicates that no chart has been found as dependency of the `{HelmUnitTestTestTarget.alias}` or	×
67	`{HelmUnitTestTestsGeneratorTarget.alias}` targets."""
68
UNCOV 69	def __init__(self, address: Address) -> None:	×
70	super().__init__(	×
71	f"No valid `{HelmChartTarget.alias}` target has been found as a dependency for target at '{address.spec}'."
72	)
73
74
UNCOV 75	@dataclass(frozen=True)	×
UNCOV 76	class HelmUnitTestFieldSet(TestFieldSet, GenerateSnapshotsFieldSet):	×
UNCOV 77	required_fields = (HelmUnitTestSourceField,)	×
78
UNCOV 79	source: HelmUnitTestSourceField	×
UNCOV 80	dependencies: HelmUnitTestDependenciesField	×
UNCOV 81	strict: HelmUnitTestStrictField	×
UNCOV 82	timeout: HelmUnitTestTimeoutField	×
83
84
UNCOV 85	class HelmUnitTestRequest(TestRequest):	×
NEW 86	tool_subsystem = HelmUnitTestSubsystem # type: ignore[assignment]	×
UNCOV 87	field_set_type = HelmUnitTestFieldSet	×
88
89
UNCOV 90	@dataclass(frozen=True)	×
UNCOV 91	class HelmUnitTestSetup:	×
UNCOV 92	chart: HelmChart	×
UNCOV 93	chart_root: HelmChartRoot	×
UNCOV 94	process: HelmProcess	×
UNCOV 95	reports_output_directory: str	×
UNCOV 96	snapshot_output_directories: tuple[str, ...]	×
97
98
UNCOV 99	@dataclass(frozen=True)	×
UNCOV 100	class HelmUnitTestSetupRequest:	×
UNCOV 101	field_set: HelmUnitTestFieldSet	×
UNCOV 102	description: str = dataclasses.field(compare=False)	×
UNCOV 103	force: bool	×
UNCOV 104	update_snapshots: bool	×
UNCOV 105	timeout_seconds: int \| None	×
106
107
UNCOV 108	@rule	×
UNCOV 109	async def setup_helm_unittest(	×
110	request: HelmUnitTestSetupRequest, unittest_subsystem: HelmUnitTestSubsystem
111	) -> HelmUnitTestSetup:
112	field_set = request.field_set	×
113	direct_dep_targets, transitive_targets = await concurrently(	×
114	resolve_targets(**implicitly(DependenciesRequest(field_set.dependencies))),
115	transitive_targets_get(TransitiveTargetsRequest([field_set.address]), **implicitly()),
116	)
117	chart_targets = [tgt for tgt in direct_dep_targets if HelmChartFieldSet.is_applicable(tgt)]	×
118	if len(chart_targets) == 0:	×
119	raise MissingUnitTestChartDependency(field_set.address)	×
120
121	chart_target = chart_targets[0]	×
122	chart, chart_root, test_files, extra_files = await concurrently(	×
123	get_helm_chart(HelmChartRequest.from_target(chart_target), **implicitly()),
124	find_chart_source_root(HelmChartRootRequest(chart_target[HelmChartMetaSourceField])),
125	determine_source_files(SourceFilesRequest(sources_fields=[field_set.source])),
126	strip_source_roots(
127	**implicitly(
128	SourceFilesRequest(
129	sources_fields=[
130	tgt.get(SourcesField) for tgt in transitive_targets.dependencies
131	],
132	for_sources_types=(ResourceSourceField, FileSourceField),
133	enable_codegen=True,
134	)
135	)
136	),
137	)
138
139	stripped_test_files = await remove_prefix(	×
140	RemovePrefix(test_files.snapshot.digest, chart_root.path)
141	)
142	merged_digests = await merge_digests(	×
143	MergeDigests(
144	[
145	chart.snapshot.digest,
146	stripped_test_files,
147	extra_files.snapshot.digest,
148	]
149	)
150	)
151	input_digest = await add_prefix(AddPrefix(merged_digests, chart.name))	×
152
153	reports_dir = "__reports_dir"	×
154	reports_file = os.path.join(reports_dir, f"{field_set.address.path_safe_spec}.xml")	×
155
156	snapshot_dirs = {	×
157	os.path.join(
158	chart.name, os.path.relpath(os.path.dirname(file), chart_root.path), "__snapshot__"
159	)
160	for file in test_files.snapshot.files
161	}
162
163	# Cache test runs only if they are successful, or not at all if `--test-force`.
164	cache_scope = ProcessCacheScope.PER_SESSION if request.force else ProcessCacheScope.SUCCESSFUL	×
165
166	process = HelmProcess(	×
167	argv=[
168	unittest_subsystem.plugin_name,
169	# Always include colors and strip them out for display below (if required), for better cache
170	# hit rates
171	"--color",
172	*(("--strict",) if field_set.strict.value else ()),
173	*(("--update-snapshot",) if request.update_snapshots else ()),
174	"--output-type",
175	unittest_subsystem.output_type.value,
176	"--output-file",
177	reports_file,
178	chart.name,
179	],
180	description=request.description,
181	input_digest=input_digest,
182	cache_scope=cache_scope,
183	timeout_seconds=request.timeout_seconds if request.timeout_seconds else None,
184	output_directories=(reports_dir, *((snapshot_dirs) if request.update_snapshots else ())),
185	)
186
187	return HelmUnitTestSetup(	×
188	chart,
189	chart_root,
190	process,
191	reports_output_directory=reports_dir,
192	snapshot_output_directories=tuple(snapshot_dirs),
193	)
194
195
UNCOV 196	@rule(desc="Run Helm Unittest", level=LogLevel.DEBUG)	×
UNCOV 197	async def run_helm_unittest(	×
198	batch: HelmUnitTestRequest.Batch[HelmUnitTestFieldSet, Any],
199	test_subsystem: TestSubsystem,
200	unittest_subsystem: HelmUnitTestSubsystem,
201	) -> TestResult:
202	field_set = batch.single_element	×
203
204	setup = await setup_helm_unittest(	×
205	HelmUnitTestSetupRequest(
206	field_set,
207	description=f"Running Helm unittest suite {field_set.address}",
208	force=test_subsystem.force,
209	update_snapshots=False,
210	timeout_seconds=field_set.timeout.calculate_from_global_options(test_subsystem),
211	),
212	**implicitly(),
213	)
214	process = await helm_process(setup.process, **implicitly())	×
215	process_results = await execute_process_with_retry(	×
216	ProcessWithRetries(process, test_subsystem.attempts_default)
217	)
218
219	reports_digest = await digest_subset_to_digest(	×
220	DigestSubset(
221	process_results.last.output_digest,
222	PathGlobs([os.path.join(setup.reports_output_directory, "**")]),
223	)
224	)
225	reports = await digest_to_snapshot(	×
226	**implicitly(RemovePrefix(reports_digest, setup.reports_output_directory))
227	)
228
229	return TestResult.from_fallible_process_result(	×
230	process_results=process_results.results,
231	address=field_set.address,
232	output_setting=test_subsystem.output,
233	xml_results=reports,
234	output_simplifier=Simplifier(strip_formatting=not unittest_subsystem.color),
235	)
236
237
UNCOV 238	@rule	×
UNCOV 239	async def generate_helm_unittest_snapshots(	×
240	field_set: HelmUnitTestFieldSet,
241	) -> GenerateSnapshotsResult:
242	setup = await setup_helm_unittest(	×
243	HelmUnitTestSetupRequest(
244	field_set,
245	description=f"Generating Helm unittest snapshots for suite {field_set.address}",
246	force=False,
247	update_snapshots=True,
248	timeout_seconds=None,
249	),
250	**implicitly(),
251	)
252
253	process_result = await fallible_to_exec_result_or_raise(	×
254	**implicitly({setup.process: HelmProcess})
255	)
256	snapshot_output_digest = await digest_subset_to_digest(	×
257	DigestSubset(
258	process_result.output_digest,
259	PathGlobs(
260	[
261	os.path.join(snapshot_path, "*.snap")
262	for snapshot_path in setup.snapshot_output_directories
263	]
264	),
265	)
266	)
267
268	stripped_test_snapshot_output = await remove_prefix(	×
269	RemovePrefix(snapshot_output_digest, setup.chart.name)
270	)
271	normalised_test_snapshots = await digest_to_snapshot(	×
272	**implicitly(AddPrefix(stripped_test_snapshot_output, setup.chart_root.path))
273	)
274	return GenerateSnapshotsResult(normalised_test_snapshots)	×
275
276
UNCOV 277	def rules():	×
UNCOV 278	return [	×
279	*collect_rules(),
280	*subsystem_rules(),
281	*dependency_rules(),
282	*tool.rules(),
283	*HelmUnitTestRequest.rules(),
284	UnionRule(GenerateSnapshotsFieldSet, HelmUnitTestFieldSet),
285	]

pantsbuild / pants / 18913303678

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